From cf056b39e3604154b704c273ef091dcac2e3b2bd Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 3 Jul 2026 11:37:02 +0800 Subject: [PATCH] fix(core-storage): fix critical correctness defects from core-storage reliability audit (#4222) * fix(core-storage): fix critical correctness defects from core-storage audit Fixes verified defects found in a deep audit of the core storage path (erasure coding, disk persistence, quorum, heal, replication resync): - ecstore/disk: rewrite live xl.meta atomically (temp+rename) in delete_versions_internal and write_metadata instead of in-place truncate, which exposed torn metadata to concurrent readers and crashes on the DeleteObjects hot path - ecstore/erasure: allow heal to reconstruct from exactly data_shards bitrot-verified sources; requiring data_shards+1 made objects permanently unhealable after losing parity_shards disks - ecstore/set_disk: direct-memory inline GET applied the erasure distribution permutation twice (shuffled inputs re-indexed through distribution), concatenating wrong shards into the response body in degraded reads; collect from canonical disk-ordered inputs - ecstore/set_disk: heal now preserves the committed inline layout instead of recomputing it with a hardcoded unversioned threshold, which split quorum identity of healed replicas and caused endless re-heal churn - ecstore/replication: resync results channel switched from broadcast(1) to mpsc; a lagged broadcast receiver ended the stats collector and every subsequent failure went uncounted, letting failed resyncs be marked completed - ecstore/replication: ignore an empty persisted resync checkpoint; resuming with one skipped every object and marked the resync completed without replicating anything - ecstore/replication: fix inverted not-found error classification in replicate_object/replicate_delete logging paths - ecstore/erasure: guard decode paths against zero block_size or data_shards from corrupt on-disk metadata (divide-by-zero panic) - ecstore/disk: os::read_dir no longer consumes the entry limit on entries it does not return (is_empty_dir misjudgment); create_file opens with O_TRUNC to avoid stale trailing bytes - filemeta: treat Some(nil) version id as a null version in matches_not_strict; disk-loaded headers never store None, so the mod_time quorum guard for unversioned overwrites never fired and an interrupted overwrite could displace the committed version in merge - filemeta: fix msgpack skip lengths for fixext (missed the ext type byte) and ext16/32 (over-skipped) unknown fields - filemeta: return FileCorrupt instead of usize underflow when xl.meta is truncated inside the CRC trailer - filemeta: surface delete-marker insertion failure in delete_version instead of reporting success when the data dir is shared Co-Authored-By: Claude Fable 5 * fix(replication): drop duplicate cfg(test) etag import from boundary module The test module already imports content_matches_by_etag locally, so the top-level cfg(test) import is unused under -D warnings and fails clippy. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- .../replication/replication_resyncer.rs | 28 +++++++++++-------- crates/ecstore/src/disk/local.rs | 15 ++++++---- crates/ecstore/src/disk/os.rs | 4 +++ crates/ecstore/src/erasure/coding/decode.rs | 7 +++++ crates/ecstore/src/erasure/coding/heal.rs | 20 ++++--------- crates/ecstore/src/set_disk/heal.rs | 9 +++--- crates/ecstore/src/set_disk/read.rs | 17 +++++++++-- crates/filemeta/src/filemeta.rs | 11 +++++--- crates/filemeta/src/filemeta/msgp_decode.rs | 18 ++++++------ crates/filemeta/src/filemeta/version.rs | 14 +++++++--- 10 files changed, 89 insertions(+), 54 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 9c10ef078..d1736249c 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -101,7 +101,7 @@ static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new(); async fn finish_resync_workers( worker_txs: Vec>, - results_tx: tokio::sync::broadcast::Sender, + results_tx: tokio::sync::mpsc::Sender, futures: Vec>, abort: bool, ) -> bool { @@ -617,8 +617,12 @@ impl ReplicationResyncer { .unwrap_or_default() }; - let mut last_checkpoint = if status.resync_status == ResyncStatusType::ResyncStarted - || status.resync_status == ResyncStatusType::ResyncFailed + // An empty checkpoint means no per-object progress was persisted before the + // interruption: resume from the beginning, otherwise `object.name != checkpoint` + // below would skip every object and mark the resync completed without work. + let mut last_checkpoint = if (status.resync_status == ResyncStatusType::ResyncStarted + || status.resync_status == ResyncStatusType::ResyncFailed) + && !status.object.is_empty() { Some(status.object) } else { @@ -626,7 +630,9 @@ impl ReplicationResyncer { }; let mut worker_txs = Vec::new(); - let (results_tx, mut results_rx) = tokio::sync::broadcast::channel::(1); + // mpsc, not broadcast: a lagging broadcast receiver returns Err(Lagged) which + // would end the collector and silently drop every subsequent worker result. + let (results_tx, mut results_rx) = tokio::sync::mpsc::channel::(RESYNC_WORKER_COUNT * 4); let opts_clone = opts.clone(); let self_clone = self.clone(); @@ -634,7 +640,7 @@ impl ReplicationResyncer { let mut futures = Vec::new(); let results_fut = tokio::spawn(async move { - while let Ok(st) = results_rx.recv().await { + while let Some(st) = results_rx.recv().await { self_clone.inc_stats(&st, opts_clone.clone()).await; } }); @@ -746,7 +752,7 @@ impl ReplicationResyncer { return; } - if let Err(err) = results_tx.send(st) { + if let Err(err) = results_tx.send(st).await { error!( event = EVENT_RESYNC_RUNTIME_CHANNEL_FAILED, component = LOG_COMPONENT_ECSTORE, @@ -2128,7 +2134,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { { Ok(gr) => gr, Err(e) => { - if !is_err_object_not_found(&e) || is_err_version_not_found(&e) { + if !(is_err_object_not_found(&e) || is_err_version_not_found(&e)) { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2414,7 +2420,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { { Ok(gr) => gr, Err(e) => { - if !is_err_object_not_found(&e) || is_err_version_not_found(&e) { + if !(is_err_object_not_found(&e) || is_err_version_not_found(&e)) { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -3376,9 +3382,9 @@ mod tests { #[tokio::test] async fn test_finish_resync_workers_closes_result_collector() { let (worker_tx, mut worker_rx) = tokio::sync::mpsc::channel::(1); - let (results_tx, mut results_rx) = tokio::sync::broadcast::channel::(1); + let (results_tx, mut results_rx) = tokio::sync::mpsc::channel::(1); let worker = tokio::spawn(async move { while worker_rx.recv().await.is_some() {} }); - let collector = tokio::spawn(async move { while results_rx.recv().await.is_ok() {} }); + let collector = tokio::spawn(async move { while results_rx.recv().await.is_some() {} }); let failed = tokio::time::timeout( TokioDuration::from_secs(1), @@ -3392,7 +3398,7 @@ mod tests { #[tokio::test] async fn test_finish_resync_workers_reports_join_failure() { - let (results_tx, _results_rx) = tokio::sync::broadcast::channel::(1); + let (results_tx, _results_rx) = tokio::sync::mpsc::channel::(1); let failed_worker = tokio::spawn(async { panic!("intentional resync worker failure"); }); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index fff424ef8..1150f989f 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -1623,12 +1623,11 @@ impl LocalDisk { return Ok(()); } - // Update xl.meta + // Update xl.meta atomically: a concurrent reader or crash mid-write must + // never observe a truncated xl.meta for versions that were not deleted. let buf = fm.marshal_msg()?; - let volume_dir = self.get_bucket_path(volume)?; - - self.write_all_private(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), buf.into(), true, &volume_dir) + self.write_all_meta(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), &buf, true) .await?; Ok(()) @@ -2873,7 +2872,9 @@ impl DiskAPI for LocalDisk { if let Some(parent) = file_path.parent() { os::make_dir_all(parent, &volume_dir).await?; } - let f = super::fs::open_file(&file_path, O_CREATE | O_WRONLY) + // O_TRUNC: if a file already exists at this path, stale trailing bytes past + // the new content would otherwise survive and mismatch the metadata size. + let f = super::fs::open_file(&file_path, O_CREATE | O_WRONLY | O_TRUNC) .await .map_err(to_file_error)?; let reclaim_on_shutdown = should_reclaim_file_cache_after_write(_file_size); @@ -3927,7 +3928,9 @@ impl DiskAPI for LocalDisk { let fm_data = meta.marshal_msg()?; - self.write_all(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), fm_data.into()) + // Atomic temp+rename: this path also rewrites live xl.meta (delete markers, + // decommission), where an in-place truncate would expose torn metadata. + self.write_all_meta(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str(), &fm_data, true) .await?; Ok(()) diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 4fffcffa1..793295bfd 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -117,6 +117,10 @@ pub async fn read_dir(path: impl AsRef, count: i32) -> std::io::Result 0 { - self.data_shards + 1 - } else { - self.data_shards - }; - if available_shards < required_shards { - return Err(Error::other(format!( - "can not reconstruct data: not enough verified heal source shards (need {}, have {}) {errs:?}", - required_shards, available_shards - ))); - } - let source_parity = shards .iter() .enumerate() diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index f242463e5..cbbdf5094 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/heal.rs @@ -441,10 +441,11 @@ impl SetDisks { } } - let is_inline_buffer = runtime_sources::storage_class_should_inline( - erasure.shard_file_size(latest_meta.size), - false, - ); + // Preserve the committed layout: recomputing inline-ness here + // (with a hardcoded unversioned threshold) makes healed replicas + // diverge from healthy ones in quorum identity, so heal would + // flag them forever. + let is_inline_buffer = latest_meta.inline_data(); // create writers for all disk positions, but only for outdated disks for (index, disk_op) in out_dated_disks.iter().enumerate() { if let Some(outdated_disk) = disk_op { diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 77f205d34..65656867d 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -2527,7 +2527,6 @@ impl SetDisks { return Ok(None); } - let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); let checksum_info = fi.erasure.get_checksum_info(part.number); let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { HashAlgorithm::HighwayHash256SLegacy @@ -2537,7 +2536,11 @@ impl SetDisks { let read_length = erasure.shard_file_offset(0, object_size, object_size); if fi.data.is_some() { - let Some(data_files) = collect_inline_data_shard_fileinfos_by_index(&files, fi, erasure.data_shards, |index| { + // Collect from the canonical disk-ordered inputs: the helper indexes + // fi.erasure.distribution by disk position, so passing shard-ordered + // (shuffled) arrays would apply the permutation twice and concatenate + // the wrong shards into the response body. + let Some(data_files) = collect_inline_data_shard_fileinfos_by_index(files, fi, erasure.data_shards, |index| { disks.get(index).is_some_and(Option::is_some) }) else { return Ok(None); @@ -2587,6 +2590,7 @@ impl SetDisks { return Ok(body); } + let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); let use_mmap_read = object_mmap_read_enabled(); let reader_setup_stage_start = Instant::now(); @@ -2746,6 +2750,15 @@ impl SetDisks { fi.uses_legacy_checksum, ); + // Erasure params come from on-disk metadata; zero values must fail the read + // instead of panicking on the block/shard divisions below. + if erasure.block_size == 0 || erasure.data_shards == 0 { + return Err(Error::other(format!( + "invalid erasure metadata for {bucket}/{object}: block_size={}, data_blocks={}", + erasure.block_size, erasure.data_shards + ))); + } + let part_indices: Vec = (part_index..=last_part_index).collect(); debug!(bucket, object, ?part_indices, "Multipart part indices to stream"); diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index e486be27f..79a83b015 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/crates/filemeta/src/filemeta.rs @@ -665,14 +665,17 @@ impl FileMeta { err = self.add_version_filemata(ventry).err(); } - if self.shared_data_dir_count(obj_version_id, obj_data_dir) > 0 { - return Ok(None); - } - + // A failed delete-marker insertion must surface even when the data dir is + // shared: reporting success here silently turns the delete into a permanent + // delete that replication never propagates. if let Some(e) = err { return Err(e); } + if self.shared_data_dir_count(obj_version_id, obj_data_dir) > 0 { + return Ok(None); + } + Ok(obj_data_dir) } diff --git a/crates/filemeta/src/filemeta/msgp_decode.rs b/crates/filemeta/src/filemeta/msgp_decode.rs index 68e44ca02..b7f9606b6 100644 --- a/crates/filemeta/src/filemeta/msgp_decode.rs +++ b/crates/filemeta/src/filemeta/msgp_decode.rs @@ -179,28 +179,30 @@ pub(crate) fn skip_msgp_value(rd: &mut R) -> Result<()> { } return Ok(()); } - Marker::FixExt1 => 1, - Marker::FixExt2 => 2, - Marker::FixExt4 => 4, - Marker::FixExt8 => 8, - Marker::FixExt16 => 16, + // fixext N = marker + 1 type byte + N data bytes + Marker::FixExt1 => 2, + Marker::FixExt2 => 3, + Marker::FixExt4 => 5, + Marker::FixExt8 => 9, + Marker::FixExt16 => 17, + // ext 8/16/32 = marker + length bytes (read here) + 1 type byte + data Marker::Ext8 => { let mut b = [0u8; 1]; rd.read_exact(&mut b).map_err(Error::from)?; let len = b[0] as usize; - 1 + len // type byte + data + 1 + len } Marker::Ext16 => { let mut b = [0u8; 2]; rd.read_exact(&mut b).map_err(Error::from)?; let len = u16::from_be_bytes(b) as usize; - 2 + len // type bytes + data + 1 + len } Marker::Ext32 => { let mut b = [0u8; 4]; rd.read_exact(&mut b).map_err(Error::from)?; let len = u32::from_be_bytes(b) as usize; - 4 + len // type bytes + data + 1 + len } Marker::Reserved => 0, }; diff --git a/crates/filemeta/src/filemeta/version.rs b/crates/filemeta/src/filemeta/version.rs index d33be0d2f..44ef9e25a 100644 --- a/crates/filemeta/src/filemeta/version.rs +++ b/crates/filemeta/src/filemeta/version.rs @@ -826,7 +826,10 @@ impl FileMetaVersionHeader { pub fn matches_not_strict(&self, o: &FileMetaVersionHeader) -> bool { let mut ok = self.version_id == o.version_id && self.version_type == o.version_type && self.matches_ec(o); - if self.version_id.is_none() { + // Disk-loaded headers keep the null version as Some(nil), not None: all null + // versions share one id, so mod_time is the only thing distinguishing an + // interrupted overwrite from the committed version. + if self.version_id.is_none() || self.version_id == Some(Uuid::nil()) { ok = ok && self.mod_time == o.mod_time; } @@ -2916,11 +2919,14 @@ pub async fn read_xl_meta_no_data(reader: &mut R, size: us return Err(Error::FileCorrupt); } - let tmp = &buf[want..]; + // The metadata block is followed by a 5-byte msgp uint32 CRC trailer; + // a file truncated inside the trailer is corrupt, not a shorter meta. let crc_size = 5; - let other_size = tmp.len() - crc_size; + if buf.len() - want < crc_size { + return Err(Error::FileCorrupt); + } - want += tmp.len() - other_size; + want += crc_size; buf.truncate(want); Ok(buf)