feat(replication): add dormant MRF v2 reader (#5672)

crates/replication/src/lib.rs

@@ -48,8 +48,9 @@ pub use filemeta::{
     replication_statuses_map, target_reset_header, version_purge_statuses_map,
 };
 pub use mrf::{
-    MrfCapabilities, MrfCapability, MrfEnvelope, MrfEnvelopeError, MrfOpKind, MrfProtocolCapabilities, MrfReplicateEntry,
-    decode_mrf_file, encode_mrf_file,
+    MRF_ENVELOPE_FORMAT, MRF_ENVELOPE_VERSION, MRF_V2_FILE, MRF_V2_FORMAT, MRF_V2_NAMESPACE, MRF_V2_VERSION, MrfCapabilities,
+    MrfCapability, MrfEnvelope, MrfEnvelopeError, MrfOpKind, MrfProtocolCapabilities, MrfReplicateEntry, MrfV2Capabilities,
+    MrfV2Envelope, MrfV2Error, MrfV2Reader, MrfV2Readiness, decode_mrf_file, encode_mrf_file,
 };
 pub use multipart::{
     ReplicationMultipartPartInput, ReplicationMultipartPartPlan, ReplicationMultipartPlanError, ReplicationMultipartRange,

crates/replication/src/mrf.rs

@@ -340,6 +340,234 @@ impl fmt::Display for MrfEnvelopeError {
 
 impl std::error::Error for MrfEnvelopeError {}
 
+pub const MRF_V2_NAMESPACE: &str = "config/replication-v2";
+pub const MRF_V2_FILE: &str = "config/replication-v2/mrf.bin";
+pub const MRF_V2_FORMAT: u16 = 2;
+pub const MRF_V2_VERSION: u16 = 2;
+
+const MRF_V2_MAGIC: [u8; 4] = *b"MRF2";
+const MRF_V2_HEADER_LEN: usize = 24;
+const MRF_V2_KNOWN_CAPABILITIES: u64 = 0b1111;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum MrfV2Error {
+    InvalidMagic,
+    Truncated,
+    UnsupportedNamespace { namespace: String },
+    UnsupportedFormat { format: u16 },
+    UnsupportedVersion { version: u16 },
+    InvalidVersionRange { version: u16, min_reader_version: u16 },
+    RollbackFenced { min_reader_version: u16, reader_version: u16 },
+    ReservedHeaderBits { bits: u16 },
+    UnknownCapabilities { bits: u64 },
+    MissingCapabilities { required: u64, available: u64 },
+    PayloadLengthMismatch { declared: u32, actual: usize },
+    WriterEnabled,
+}
+
+impl std::fmt::Display for MrfV2Error {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::InvalidMagic => write!(f, "invalid MRF v2 magic"),
+            Self::Truncated => write!(f, "truncated MRF v2 envelope"),
+            Self::UnsupportedNamespace { namespace } => write!(f, "unsupported MRF v2 namespace {namespace}"),
+            Self::UnsupportedFormat { format } => write!(f, "unsupported MRF v2 format {format}"),
+            Self::UnsupportedVersion { version } => write!(f, "unsupported MRF v2 version {version}"),
+            Self::InvalidVersionRange {
+                version,
+                min_reader_version,
+            } => {
+                write!(f, "invalid MRF v2 version range: version {version}, minimum reader {min_reader_version}")
+            }
+            Self::RollbackFenced {
+                min_reader_version,
+                reader_version,
+            } => write!(
+                f,
+                "MRF v2 rollback fenced: reader version {reader_version} is below required {min_reader_version}"
+            ),
+            Self::ReservedHeaderBits { bits } => write!(f, "reserved MRF v2 header bits are set: 0x{bits:04x}"),
+            Self::UnknownCapabilities { bits } => write!(f, "unknown MRF v2 capability bits 0x{bits:016x}"),
+            Self::MissingCapabilities { required, available } => {
+                write!(f, "MRF v2 capabilities 0x{available:016x} do not satisfy required 0x{required:016x}")
+            }
+            Self::PayloadLengthMismatch { declared, actual } => {
+                write!(f, "MRF v2 payload is {actual} bytes, expected {declared}")
+            }
+            Self::WriterEnabled => write!(f, "MRF v2 writer must remain dormant"),
+        }
+    }
+}
+
+impl std::error::Error for MrfV2Error {}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct MrfV2Capabilities(u64);
+
+impl MrfV2Capabilities {
+    pub const fn current() -> Self {
+        Self(MRF_V2_KNOWN_CAPABILITIES)
+    }
+
+    pub const fn empty() -> Self {
+        Self(0)
+    }
+
+    pub const fn bits(self) -> u64 {
+        self.0
+    }
+
+    pub fn from_bits(bits: u64) -> std::result::Result<Self, MrfV2Error> {
+        let unknown = bits & !MRF_V2_KNOWN_CAPABILITIES;
+        if unknown != 0 {
+            return Err(MrfV2Error::UnknownCapabilities { bits: unknown });
+        }
+        Ok(Self(bits))
+    }
+
+    pub const fn supports(self, required: Self) -> bool {
+        self.0 & required.0 == required.0
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct MrfV2Readiness {
+    reader_version: u16,
+    capabilities: MrfV2Capabilities,
+    writer_enabled: bool,
+}
+
+impl MrfV2Readiness {
+    pub const fn dormant() -> Self {
+        Self {
+            reader_version: MRF_V2_VERSION,
+            capabilities: MrfV2Capabilities::current(),
+            writer_enabled: false,
+        }
+    }
+
+    pub const fn new(reader_version: u16, capabilities: MrfV2Capabilities, writer_enabled: bool) -> Self {
+        Self {
+            reader_version,
+            capabilities,
+            writer_enabled,
+        }
+    }
+
+    pub const fn writer_enabled(self) -> bool {
+        self.writer_enabled
+    }
+
+    pub fn reader(self) -> std::result::Result<MrfV2Reader, MrfV2Error> {
+        if self.writer_enabled {
+            return Err(MrfV2Error::WriterEnabled);
+        }
+        if self.reader_version != MRF_V2_VERSION {
+            return Err(MrfV2Error::UnsupportedVersion {
+                version: self.reader_version,
+            });
+        }
+        MrfV2Capabilities::from_bits(self.capabilities.bits())?;
+        Ok(MrfV2Reader { readiness: self })
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct MrfV2Reader {
+    readiness: MrfV2Readiness,
+}
+
+impl MrfV2Reader {
+    pub fn read(self, namespace: &str, data: &[u8]) -> std::result::Result<MrfV2Envelope, MrfV2Error> {
+        if namespace != MRF_V2_NAMESPACE {
+            return Err(MrfV2Error::UnsupportedNamespace {
+                namespace: namespace.to_string(),
+            });
+        }
+        if data.len() < MRF_V2_HEADER_LEN {
+            return Err(MrfV2Error::Truncated);
+        }
+        if data[..4] != MRF_V2_MAGIC {
+            return Err(MrfV2Error::InvalidMagic);
+        }
+        let format = LittleEndian::read_u16(&data[4..6]);
+        if format != MRF_V2_FORMAT {
+            return Err(MrfV2Error::UnsupportedFormat { format });
+        }
+        let version = LittleEndian::read_u16(&data[6..8]);
+        if version != MRF_V2_VERSION || version != self.readiness.reader_version {
+            return Err(MrfV2Error::UnsupportedVersion { version });
+        }
+        let min_reader_version = LittleEndian::read_u16(&data[8..10]);
+        if min_reader_version > self.readiness.reader_version {
+            return Err(MrfV2Error::RollbackFenced {
+                min_reader_version,
+                reader_version: self.readiness.reader_version,
+            });
+        }
+        if min_reader_version > version {
+            return Err(MrfV2Error::InvalidVersionRange {
+                version,
+                min_reader_version,
+            });
+        }
+        let reserved = LittleEndian::read_u16(&data[10..12]);
+        if reserved != 0 {
+            return Err(MrfV2Error::ReservedHeaderBits { bits: reserved });
+        }
+        let capabilities = MrfV2Capabilities::from_bits(LittleEndian::read_u64(&data[12..20]))?;
+        if !self.readiness.capabilities.supports(capabilities) {
+            return Err(MrfV2Error::MissingCapabilities {
+                required: capabilities.bits(),
+                available: self.readiness.capabilities.bits(),
+            });
+        }
+        let payload_len = LittleEndian::read_u32(&data[20..24]);
+        let actual = data.len() - MRF_V2_HEADER_LEN;
+        if usize::try_from(payload_len).map_err(|_| MrfV2Error::PayloadLengthMismatch {
+            declared: payload_len,
+            actual,
+        })? != actual
+        {
+            return Err(MrfV2Error::PayloadLengthMismatch {
+                declared: payload_len,
+                actual,
+            });
+        }
+        Ok(MrfV2Envelope {
+            version,
+            min_reader_version,
+            capabilities,
+            payload: data[MRF_V2_HEADER_LEN..].to_vec(),
+        })
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct MrfV2Envelope {
+    version: u16,
+    min_reader_version: u16,
+    capabilities: MrfV2Capabilities,
+    payload: Vec<u8>,
+}
+
+impl MrfV2Envelope {
+    pub const fn version(&self) -> u16 {
+        self.version
+    }
+
+    pub const fn min_reader_version(&self) -> u16 {
+        self.min_reader_version
+    }
+
+    pub const fn capabilities(&self) -> MrfV2Capabilities {
+        self.capabilities
+    }
+
+    pub fn payload(&self) -> &[u8] {
+        &self.payload
+    }
+}
 pub fn encode_mrf_file(entries: &[MrfReplicateEntry]) -> Result<Vec<u8>> {
     let payload = rmp_serde::to_vec_named(entries).map_err(|e| Error::Other(e.to_string()))?;
     let mut data = Vec::with_capacity(4 + payload.len());
@@ -653,4 +881,87 @@ mod tests {
         assert_eq!(negotiated.version(), 1);
         assert_eq!(negotiated.min_reader_version(), 1);
     }
+
+    #[test]
+    fn v1_file_contract_and_v2_namespace_remain_separate() {
+        assert_eq!(encode_mrf_file(&[]).expect("v1 empty file should encode"), vec![1, 0, 1, 0, 0x90]);
+        assert_eq!(MRF_V2_FILE, "config/replication-v2/mrf.bin");
+        assert_ne!(MRF_V2_FILE, "config/replication/mrf.bin");
+    }
+
+    #[test]
+    fn dormant_v2_reader_accepts_stable_fixture() {
+        let fixture = [
+            b'M', b'R', b'F', b'2', 2, 0, 2, 0, 2, 0, 0, 0, 15, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
+        ];
+        let readiness = MrfV2Readiness::dormant();
+        assert!(!readiness.writer_enabled());
+        let envelope = readiness
+            .reader()
+            .expect("dormant readiness should expose the reader")
+            .read(MRF_V2_NAMESPACE, &fixture)
+            .expect("v2 fixture should decode");
+        assert_eq!(envelope.version(), MRF_V2_VERSION);
+        assert_eq!(envelope.min_reader_version(), MRF_V2_VERSION);
+        assert_eq!(envelope.capabilities(), MrfV2Capabilities::current());
+        assert_eq!(envelope.payload(), &[1, 2, 3]);
+    }
+
+    #[test]
+    fn v2_reader_rejects_wrong_namespace_version_capability_and_rollback() {
+        let fixture = [
+            b'M', b'R', b'F', b'2', 2, 0, 2, 0, 2, 0, 0, 0, 15, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
+        ];
+        let reader = MrfV2Readiness::dormant().reader().expect("dormant reader should initialize");
+        assert!(matches!(
+            reader.read("config/replication/mrf.bin", &fixture),
+            Err(MrfV2Error::UnsupportedNamespace { .. })
+        ));
+
+        let mut version = fixture;
+        version[6..8].copy_from_slice(&1u16.to_le_bytes());
+        assert_eq!(
+            reader.read(MRF_V2_NAMESPACE, &version),
+            Err(MrfV2Error::UnsupportedVersion { version: 1 })
+        );
+
+        let mut capabilities = fixture;
+        capabilities[12..20].copy_from_slice(&(1u64 << 63).to_le_bytes());
+        assert_eq!(
+            reader.read(MRF_V2_NAMESPACE, &capabilities),
+            Err(MrfV2Error::UnknownCapabilities { bits: 1u64 << 63 })
+        );
+
+        let mut rollback = fixture;
+        rollback[8..10].copy_from_slice(&3u16.to_le_bytes());
+        assert_eq!(
+            reader.read(MRF_V2_NAMESPACE, &rollback),
+            Err(MrfV2Error::RollbackFenced {
+                min_reader_version: 3,
+                reader_version: 2,
+            })
+        );
+    }
+
+    #[test]
+    fn v2_readiness_rejects_writer_enablement_and_missing_capabilities() {
+        assert_eq!(
+            MrfV2Readiness::new(2, MrfV2Capabilities::current(), true).reader(),
+            Err(MrfV2Error::WriterEnabled)
+        );
+
+        let fixture = [
+            b'M', b'R', b'F', b'2', 2, 0, 2, 0, 2, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
+        ];
+        let reader = MrfV2Readiness::new(2, MrfV2Capabilities::empty(), false)
+            .reader()
+            .expect("readiness with valid empty capabilities should initialize");
+        assert_eq!(
+            reader.read(MRF_V2_NAMESPACE, &fixture),
+            Err(MrfV2Error::MissingCapabilities {
+                required: 1,
+                available: 0,
+            })
+        );
+    }
 }