diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index 9102a5553..3638391f7 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/crates/filemeta/src/filemeta.rs @@ -24,9 +24,9 @@ use rustfs_utils::http::headers::{ AMZ_STORAGE_CLASS, }; use rustfs_utils::http::{ - AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_CRC, SUFFIX_DATA_MOV, SUFFIX_HEALING, SUFFIX_PURGESTATUS, SUFFIX_REPLICA_STATUS, - SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, has_internal_suffix, insert_bytes, - is_internal_key, + AMZ_BUCKET_REPLICATION_STATUS, MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, SUFFIX_CRC, SUFFIX_DATA_MOV, SUFFIX_HEALING, + SUFFIX_PURGESTATUS, SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_STATUS, + SUFFIX_REPLICATION_TIMESTAMP, has_internal_suffix, insert_bytes, is_internal_key, }; use s3s::header::X_AMZ_RESTORE; use serde::{Deserialize, Serialize}; @@ -108,6 +108,47 @@ pub use version::*; // type ScanHeaderVersionFn = Box Result<()>>; +/// Order two shallow versions newest-first, deriving a total order from the +/// canonical `FileMetaVersionHeader::sorts_before` predicate. +/// +/// This MUST match `sorts_before` exactly: the header-only merge path +/// (`metacache`) and latest-version selection (`version.rs`) both key off +/// `sorts_before`, so any divergence here makes different code paths disagree +/// on which version is latest when several share a `mod_time` — e.g. an object +/// and a delete marker with identical timestamps flipping between "present" and +/// "deleted" depending on which path last sorted (backlog#799 B15). +fn cmp_shallow_versions_for_order(a: &FileMetaShallowVersion, b: &FileMetaShallowVersion) -> Ordering { + if a.header.sorts_before(&b.header) { + Ordering::Less + } else if b.header.sorts_before(&a.header) { + Ordering::Greater + } else { + Ordering::Equal + } +} + +/// Persists replication reset state into `meta_sys` under BOTH internal +/// prefixes (`x-rustfs-internal-*` and `x-minio-internal-*`). +/// +/// Reset entries reach this point keyed either by a bare ARN +/// (`ObjectInfo::replication_state`) or by an already-prefixed internal key +/// (`get_internal_replication_state` / `target_reset_header`). The old code +/// inserted the key verbatim: a bare ARN was written with no internal prefix at +/// all, so read-back — which only recognizes prefixed keys — silently dropped +/// the reset state, and a rustfs-only key was invisible to MinIO-compatible +/// readers (backlog#799 B16). Normalize every entry to the canonical +/// `replication-reset-` suffix and write both prefixes. +fn persist_reset_statuses(meta_sys: &mut HashMap>, reset_statuses_map: &HashMap) { + for (k, v) in reset_statuses_map { + let suffix = k + .strip_prefix(RUSTFS_INTERNAL_PREFIX) + .or_else(|| k.strip_prefix(MINIO_INTERNAL_PREFIX)) + .map(str::to_string) + .unwrap_or_else(|| format!("{SUFFIX_REPLICATION_RESET}-{k}")); + insert_bytes(meta_sys, &suffix, v.as_bytes().to_vec()); + } +} + #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] pub struct FileMeta { pub versions: Vec, @@ -157,19 +198,7 @@ impl FileMeta { return; } - self.versions.sort_by(|a, b| { - if a.header.mod_time != b.header.mod_time { - b.header.mod_time.cmp(&a.header.mod_time) - } else if a.header.version_type != b.header.version_type { - b.header.version_type.cmp(&a.header.version_type) - } else if a.header.version_id != b.header.version_id { - b.header.version_id.cmp(&a.header.version_id) - } else if a.header.flags != b.header.flags { - b.header.flags.cmp(&a.header.flags) - } else { - b.cmp(a) - } - }); + self.versions.sort_by(cmp_shallow_versions_for_order); } fn find_inline_data_for_version(&self, version_id: Option) -> Result>> { @@ -260,19 +289,7 @@ impl FileMeta { } } - self.versions.sort_by(|a, b| { - if a.header.mod_time != b.header.mod_time { - b.header.mod_time.cmp(&a.header.mod_time) - } else if a.header.version_type != b.header.version_type { - b.header.version_type.cmp(&a.header.version_type) - } else if a.header.version_id != b.header.version_id { - b.header.version_id.cmp(&a.header.version_id) - } else if a.header.flags != b.header.flags { - b.header.flags.cmp(&a.header.flags) - } else { - b.cmp(a) - } - }); + self.versions.sort_by(cmp_shallow_versions_for_order); Ok(()) } @@ -485,15 +502,10 @@ impl FileMeta { insert_bytes(&mut delete_marker.meta_sys, SUFFIX_PURGESTATUS, value); } - if let Some(delete_marker) = ventry.delete_marker.as_mut() { - for (k, v) in fi - .replication_state_internal - .as_ref() - .map(|v| v.reset_statuses_map.clone()) - .unwrap_or_default() - { - delete_marker.meta_sys.insert(k.clone(), v.clone().as_bytes().to_vec()); - } + if let Some(delete_marker) = ventry.delete_marker.as_mut() + && let Some(state) = fi.replication_state_internal.as_ref() + { + persist_reset_statuses(&mut delete_marker.meta_sys, &state.reset_statuses_map); } } @@ -565,13 +577,8 @@ impl FileMeta { } } - for (k, v) in fi - .replication_state_internal - .as_ref() - .map(|v| v.reset_statuses_map.clone()) - .unwrap_or_default() - { - delete_marker.meta_sys.insert(k.clone(), v.clone().as_bytes().to_vec()); + if let Some(state) = fi.replication_state_internal.as_ref() { + persist_reset_statuses(&mut delete_marker.meta_sys, &state.reset_statuses_map); } } @@ -601,13 +608,8 @@ impl FileMeta { .as_bytes() .to_vec(); insert_bytes(&mut obj.meta_sys, SUFFIX_PURGESTATUS, value); - for (k, v) in fi - .replication_state_internal - .as_ref() - .map(|v| v.reset_statuses_map.clone()) - .unwrap_or_default() - { - obj.meta_sys.insert(k.clone(), v.clone().as_bytes().to_vec()); + if let Some(state) = fi.replication_state_internal.as_ref() { + persist_reset_statuses(&mut obj.meta_sys, &state.reset_statuses_map); } } @@ -935,6 +937,93 @@ mod test { buf } + /// Regression for backlog#799 B15: `sort_by_mod_time` must order versions + /// identically to `FileMetaVersionHeader::sorts_before`. When an object and + /// a delete marker share a `mod_time`, the old tie-break sorted the delete + /// marker first (so the object looked deleted) while `sorts_before` — used + /// by the metacache merge and latest-version selection — puts the object + /// first. That divergence made different code paths disagree on latest. + #[test] + fn sort_by_mod_time_matches_sorts_before_on_equal_mod_time() { + let mod_time = Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap()); + let obj_id = Uuid::from_u128(1); + let del_id = Uuid::from_u128(2); + + let object = FileMetaShallowVersion { + header: FileMetaVersionHeader { + version_id: Some(obj_id), + mod_time, + version_type: VersionType::Object, + ..Default::default() + }, + meta: Vec::new(), + }; + let delete_marker = FileMetaShallowVersion { + header: FileMetaVersionHeader { + version_id: Some(del_id), + mod_time, + version_type: VersionType::Delete, + ..Default::default() + }, + meta: Vec::new(), + }; + + // sorts_before is the canonical predicate: object precedes the marker. + assert!(object.header.sorts_before(&delete_marker.header)); + assert!(!delete_marker.header.sorts_before(&object.header)); + + // Insert marker-first so a wrong tie-break would leave it at index 0. + let mut fm = FileMeta { + versions: vec![delete_marker, object], + ..Default::default() + }; + fm.sort_by_mod_time(); + + assert_eq!( + fm.versions[0].header.version_type, + VersionType::Object, + "object must sort before an equal-mod_time delete marker, matching sorts_before" + ); + assert_eq!(fm.versions[1].header.version_type, VersionType::Delete); + assert!(fm.is_sorted_by_mod_time()); + } + + /// Regression for backlog#799 B16: replication reset state must be + /// persisted under both internal prefixes, never as a bare ARN. A bare-ARN + /// key (produced by `ObjectInfo::replication_state`) has no internal prefix, + /// so read-back — which only recognizes prefixed keys — silently dropped it. + #[test] + fn persist_reset_statuses_normalizes_to_dual_prefixed_keys() { + let arn = "arn:rustfs:replication::target:bucket"; + let ts = "2026-06-30T00:00:00Z;reset-1"; + let suffix = format!("{SUFFIX_REPLICATION_RESET}-{arn}"); + let rustfs_key = format!("{RUSTFS_INTERNAL_PREFIX}{suffix}"); + let minio_key = format!("{MINIO_INTERNAL_PREFIX}{suffix}"); + + // Bare-ARN key must be normalized to prefixed keys, never written raw. + let mut bare = HashMap::new(); + bare.insert(arn.to_string(), ts.to_string()); + let mut meta_sys = HashMap::new(); + persist_reset_statuses(&mut meta_sys, &bare); + assert_eq!(meta_sys.get(&rustfs_key).map(Vec::as_slice), Some(ts.as_bytes())); + assert_eq!(meta_sys.get(&minio_key).map(Vec::as_slice), Some(ts.as_bytes())); + assert!( + !meta_sys.contains_key(arn), + "bare ARN key must never be persisted (it is dropped on read)" + ); + assert_eq!(meta_sys.len(), 2); + + // An already-prefixed key must land on the same canonical dual keys, + // not gain a second prefix. + let mut prefixed = HashMap::new(); + prefixed.insert(rustfs_key.clone(), ts.to_string()); + let mut meta_sys2 = HashMap::new(); + persist_reset_statuses(&mut meta_sys2, &prefixed); + assert_eq!(meta_sys2.get(&rustfs_key).map(Vec::as_slice), Some(ts.as_bytes())); + assert_eq!(meta_sys2.get(&minio_key).map(Vec::as_slice), Some(ts.as_bytes())); + assert_eq!(meta_sys2.len(), 2, "must not create a double-prefixed key"); + } + /// Regression test for rustfs/rustfs#2715: a corrupted version count in /// xl.meta must yield a decode error instead of sizing a huge allocation /// from the bogus count (which aborts the whole process). diff --git a/crates/filemeta/src/filemeta/version.rs b/crates/filemeta/src/filemeta/version.rs index 6f5ff89d2..72cc7216a 100644 --- a/crates/filemeta/src/filemeta/version.rs +++ b/crates/filemeta/src/filemeta/version.rs @@ -41,6 +41,7 @@ const MSGPACK_FIXEXT4: u8 = 0xd6; const MSGPACK_FIXEXT8: u8 = 0xd7; const MSGPACK_TIME_EXT_LEGACY: i8 = 5; const MSGPACK_TIME_EXT_OFFICIAL: i8 = -1; +const MSGPACK_TIME_LEN: u8 = 12; fn read_msgp_string(rd: &mut R) -> Result { let len = rmp::decode::read_str_len(rd)? as usize; @@ -53,6 +54,18 @@ fn read_msgp_bin(rd: &mut R) -> Result> { read_exact_vec(rd, len) } +/// Writes an `OffsetDateTime` as the ext8 / legacy (type 5, 12-byte +/// seconds+nanos) msgpack time encoding used by the V1 (Legacy) object body. +/// `read_msgp_time` decodes exactly this shape via `MSGPACK_TIME_EXT_LEGACY`. +fn write_msgp_time(wr: &mut W, t: OffsetDateTime) -> Result<()> { + wr.write_all(&[MSGPACK_EXT8, MSGPACK_TIME_LEN, MSGPACK_TIME_EXT_LEGACY as u8])?; + let mut buf = [0u8; MSGPACK_TIME_LEN as usize]; + buf[0..8].copy_from_slice(&t.unix_timestamp().to_be_bytes()); + buf[8..12].copy_from_slice(&t.nanosecond().to_be_bytes()); + wr.write_all(&buf)?; + Ok(()) +} + fn deserialize_legacy_uuid_bytes<'de, D>(deserializer: D) -> std::result::Result, D::Error> where D: serde::Deserializer<'de>, @@ -560,8 +573,11 @@ impl FileMetaVersion { } pub fn encode_to(&self, wr: &mut W) -> Result<()> { - // Variable map size: omit V2Obj/DelObj when None + // Variable map size: omit V1Obj/V2Obj/DelObj when None let mut map_len: u32 = 2; // "Type" + "v" + if self.legacy_object.is_some() { + map_len += 1; + } if self.object.is_some() { map_len += 1; } @@ -575,6 +591,13 @@ impl FileMetaVersion { rmp::encode::write_str(wr, "Type")?; rmp::encode::write_uint(wr, self.version_type.to_u8() as u64)?; + // V1Obj — legacy body must round-trip; dropping it silently corrupted + // Legacy versions on re-encode (backlog#799 B18). + if let Some(ref legacy) = self.legacy_object { + rmp::encode::write_str(wr, "V1Obj")?; + legacy.encode_to(wr)?; + } + // V2Obj if let Some(ref obj) = self.object { rmp::encode::write_str(wr, "V2Obj")?; @@ -1318,6 +1341,46 @@ impl MetaObjectV1 { Ok(()) } + /// Symmetric encoder for the V1 (Legacy) object body. Kept field-for-field + /// consistent with `decode_from` so a Legacy version round-trips through + /// `FileMetaVersion::encode_to` without losing its body (backlog#799 B18). + fn encode_to(&self, wr: &mut W) -> Result<()> { + rmp::encode::write_map_len(wr, 8)?; + + rmp::encode::write_str(wr, "Version")?; + rmp::encode::write_str(wr, &self.version)?; + + rmp::encode::write_str(wr, "Format")?; + rmp::encode::write_str(wr, &self.format)?; + + rmp::encode::write_str(wr, "Stat")?; + self.stat.encode_to(wr)?; + + rmp::encode::write_str(wr, "Erasure")?; + self.erasure.encode_to(wr)?; + + rmp::encode::write_str(wr, "Meta")?; + rmp::encode::write_map_len(wr, self.meta.len() as u32)?; + for (k, v) in &self.meta { + rmp::encode::write_str(wr, k)?; + rmp::encode::write_str(wr, v)?; + } + + rmp::encode::write_str(wr, "Parts")?; + rmp::encode::write_array_len(wr, self.parts.len() as u32)?; + for part in &self.parts { + part.encode_to(wr)?; + } + + rmp::encode::write_str(wr, "VersionID")?; + rmp::encode::write_str(wr, &self.version_id)?; + + rmp::encode::write_str(wr, "DataDir")?; + rmp::encode::write_str(wr, &self.data_dir)?; + + Ok(()) + } + fn get_signature(&self) -> [u8; 4] { let mut hasher = xxhash_rust::xxh64::Xxh64::new(XXHASH_SEED); hasher.update(self.version.as_bytes()); @@ -1400,6 +1463,35 @@ impl MetaObjectV1Stat { Ok(()) } + + fn encode_to(&self, wr: &mut W) -> Result<()> { + // `ModTime` is only written when present, mirroring `decode_from` (which + // sets it only when the field exists) so a `None` mod_time stays absent + // rather than round-tripping to `Some(UNIX_EPOCH)`. + let map_len: u32 = if self.mod_time.is_some() { 5 } else { 4 }; + rmp::encode::write_map_len(wr, map_len)?; + + rmp::encode::write_str(wr, "Size")?; + rmp::encode::write_sint(wr, self.size)?; + + if let Some(mod_time) = self.mod_time { + rmp::encode::write_str(wr, "ModTime")?; + write_msgp_time(wr, mod_time)?; + } + + rmp::encode::write_str(wr, "Name")?; + rmp::encode::write_str(wr, &self.name)?; + + rmp::encode::write_str(wr, "Dir")?; + rmp::encode::write_bool(wr, self.dir)?; + + rmp::encode::write_str(wr, "Mode")?; + // `Mode` decodes with the strict `read_u32`, which only accepts the U32 + // marker — a narrower `write_uint` marker would fail to round-trip. + rmp::encode::write_u32(wr, self.mode)?; + + Ok(()) + } } impl MetaObjectV1Erasure { @@ -1440,6 +1532,39 @@ impl MetaObjectV1Erasure { Ok(()) } + + fn encode_to(&self, wr: &mut W) -> Result<()> { + rmp::encode::write_map_len(wr, 7)?; + + rmp::encode::write_str(wr, "Algorithm")?; + rmp::encode::write_str(wr, &self.algorithm)?; + + rmp::encode::write_str(wr, "DataBlocks")?; + rmp::encode::write_sint(wr, self.data_blocks as i64)?; + + rmp::encode::write_str(wr, "ParityBlocks")?; + rmp::encode::write_sint(wr, self.parity_blocks as i64)?; + + rmp::encode::write_str(wr, "BlockSize")?; + rmp::encode::write_sint(wr, self.block_size as i64)?; + + rmp::encode::write_str(wr, "Index")?; + rmp::encode::write_sint(wr, self.index as i64)?; + + rmp::encode::write_str(wr, "Distribution")?; + rmp::encode::write_array_len(wr, self.distribution.len() as u32)?; + for v in &self.distribution { + rmp::encode::write_sint(wr, *v as i64)?; + } + + rmp::encode::write_str(wr, "Checksums")?; + rmp::encode::write_array_len(wr, self.checksums.len() as u32)?; + for checksum in &self.checksums { + checksum.encode_to(wr)?; + } + + Ok(()) + } } impl MetaObjectV1ChecksumInfo { @@ -1460,6 +1585,21 @@ impl MetaObjectV1ChecksumInfo { Ok(()) } + + fn encode_to(&self, wr: &mut W) -> Result<()> { + rmp::encode::write_map_len(wr, 3)?; + + rmp::encode::write_str(wr, "PartNumber")?; + rmp::encode::write_sint(wr, self.part_number as i64)?; + + rmp::encode::write_str(wr, "Algorithm")?; + rmp::encode::write_str(wr, &self.algorithm)?; + + rmp::encode::write_str(wr, "Hash")?; + rmp::encode::write_bin(wr, &self.hash)?; + + Ok(()) + } } impl MetaObjectV1Part { @@ -1492,6 +1632,63 @@ impl MetaObjectV1Part { Ok(()) } + + fn encode_to(&self, wr: &mut W) -> Result<()> { + // Non-optional scalars are always written; optional fields only when + // present, so a decoded part round-trips (absent stays absent). + let mut map_len: u32 = 4; // e, n, s, as + if self.mod_time.is_some() { + map_len += 1; + } + if self.index.is_some() { + map_len += 1; + } + if self.checksums.is_some() { + map_len += 1; + } + if self.error.is_some() { + map_len += 1; + } + rmp::encode::write_map_len(wr, map_len)?; + + rmp::encode::write_str(wr, "e")?; + rmp::encode::write_str(wr, &self.etag)?; + + rmp::encode::write_str(wr, "n")?; + rmp::encode::write_sint(wr, self.number as i64)?; + + rmp::encode::write_str(wr, "s")?; + rmp::encode::write_sint(wr, self.size as i64)?; + + rmp::encode::write_str(wr, "as")?; + rmp::encode::write_sint(wr, self.actual_size)?; + + if let Some(mt) = self.mod_time { + rmp::encode::write_str(wr, "mt")?; + write_msgp_time(wr, mt)?; + } + + if let Some(ref index) = self.index { + rmp::encode::write_str(wr, "i")?; + rmp::encode::write_bin(wr, index)?; + } + + if let Some(ref checksums) = self.checksums { + rmp::encode::write_str(wr, "crc")?; + rmp::encode::write_map_len(wr, checksums.len() as u32)?; + for (k, v) in checksums { + rmp::encode::write_str(wr, k)?; + rmp::encode::write_str(wr, v)?; + } + } + + if let Some(ref err) = self.error { + rmp::encode::write_str(wr, "err")?; + rmp::encode::write_str(wr, err)?; + } + + Ok(()) + } } impl From for ErasureInfo { @@ -3677,4 +3874,97 @@ mod tests { "lookup must find the round-tripped reset status" ); } + + /// Regression for backlog#799 B18: a Legacy (V1Obj) version must survive an + /// encode/decode round trip. `encode_to` used to omit the `V1Obj` field + /// entirely, silently dropping the whole legacy body on re-marshal. + #[test] + fn legacy_version_body_round_trips_through_encode() { + let mut meta = HashMap::new(); + meta.insert("content-type".to_string(), "application/octet-stream".to_string()); + + let mut crc = HashMap::new(); + crc.insert("crc32c".to_string(), "deadbeef".to_string()); + + let legacy = MetaObjectV1 { + version: "1.0.1".to_string(), + format: "xl".to_string(), + stat: MetaObjectV1Stat { + size: 4096, + mod_time: Some( + OffsetDateTime::from_unix_timestamp(1_700_000_123) + .unwrap() + .replace_nanosecond(456) + .unwrap(), + ), + name: "obj".to_string(), + dir: false, + mode: 0o644, + }, + erasure: MetaObjectV1Erasure { + algorithm: "klauspost/reedsolomon/vandermonde".to_string(), + data_blocks: 4, + parity_blocks: 2, + block_size: 10 << 20, + index: 1, + distribution: vec![1, 2, 3, 4, 5, 6], + checksums: vec![MetaObjectV1ChecksumInfo { + part_number: 1, + algorithm: "highwayhash256S".to_string(), + hash: vec![0xaa, 0xbb, 0xcc, 0xdd], + }], + }, + meta, + parts: vec![MetaObjectV1Part { + etag: "etag-1".to_string(), + number: 1, + size: 4096, + actual_size: 4096, + mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_100).unwrap()), + index: Some(Bytes::from_static(&[1, 2, 3])), + checksums: Some(crc), + error: None, + }], + version_id: "00000000-0000-0000-0000-000000000001".to_string(), + data_dir: "11111111-1111-1111-1111-111111111111".to_string(), + }; + + let version = FileMetaVersion { + version_type: VersionType::Legacy, + legacy_object: Some(legacy.clone()), + ..Default::default() + }; + + let buf = version.marshal_msg().expect("marshal must succeed"); + let mut decoded = FileMetaVersion::default(); + decoded.unmarshal_msg(&buf).expect("unmarshal must succeed"); + + assert_eq!(decoded.version_type, VersionType::Legacy); + let decoded_legacy = decoded.legacy_object.expect("legacy body must survive round trip"); + assert_eq!(decoded_legacy, legacy, "legacy body must be byte-for-byte preserved"); + } + + /// A `None` `Stat.ModTime` must stay `None` across encode/decode — the + /// encoder writes the field only when present, mirroring the decoder, so it + /// never silently becomes `Some(UNIX_EPOCH)` (backlog#799 B18 hardening). + #[test] + fn legacy_stat_none_mod_time_round_trips_as_none() { + let stat = MetaObjectV1Stat { + size: 10, + mod_time: None, + name: "n".to_string(), + dir: true, + mode: 0o755, + }; + + let mut buf = Vec::new(); + stat.encode_to(&mut buf).expect("encode must succeed"); + let mut decoded = MetaObjectV1Stat::default(); + decoded + .decode_from(&mut std::io::Cursor::new(&buf)) + .expect("decode must succeed"); + + assert_eq!(decoded, stat); + assert!(decoded.mod_time.is_none(), "absent mod_time must not become Some(UNIX_EPOCH)"); + } }