From 3ce17cd7dd7c66b62d678ce46ee76fd5715e6375 Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 3 Aug 2026 22:09:33 +0800 Subject: [PATCH] feat(replication): add dormant MRF v2 reader (#5672) --- crates/replication/src/lib.rs | 5 +- crates/replication/src/mrf.rs | 311 ++++++++++++++++++++++++++++++++++ 2 files changed, 314 insertions(+), 2 deletions(-) diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index c140b4fb9..330425303 100644 --- a/crates/replication/src/lib.rs +++ b/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, diff --git a/crates/replication/src/mrf.rs b/crates/replication/src/mrf.rs index add3b0c11..f488cf1a4 100644 --- a/crates/replication/src/mrf.rs +++ b/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 { + 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 { + 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 { + 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, +} + +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> { 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, + }) + ); + } }