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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

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<tokio::sync::mpsc::Sender<ReplicateObjectInfo>>,
-    results_tx: tokio::sync::broadcast::Sender<TargetReplicationResyncStatus>,
+    results_tx: tokio::sync::mpsc::Sender<TargetReplicationResyncStatus>,
     futures: Vec<JoinHandle<()>>,
     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::<TargetReplicationResyncStatus>(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::<TargetReplicationResyncStatus>(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::<ReplicateObjectInfo>(1);
-        let (results_tx, mut results_rx) = tokio::sync::broadcast::channel::<TargetReplicationResyncStatus>(1);
+        let (results_tx, mut results_rx) = tokio::sync::mpsc::channel::<TargetReplicationResyncStatus>(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::<TargetReplicationResyncStatus>(1);
+        let (results_tx, _results_rx) = tokio::sync::mpsc::channel::<TargetReplicationResyncStatus>(1);
         let failed_worker = tokio::spawn(async {
             panic!("intentional resync worker failure");
         });

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(())

crates/ecstore/src/disk/os.rs

@@ -117,6 +117,10 @@ pub async fn read_dir(path: impl AsRef<Path>, count: i32) -> std::io::Result<Vec
             volumes.push(name);
         } else if file_type.is_dir() {
             volumes.push(format!("{name}{SLASH_SEPARATOR}"));
+        } else {
+            // Entries we don't return (symlinks, sockets, fifos) must not consume
+            // the limit: is_empty_dir/list_dir(count=1) would misreport otherwise.
+            continue;
         }
         count -= 1;
         if count == 0 {

crates/ecstore/src/erasure/coding/decode.rs

@@ -1130,6 +1130,13 @@ impl Erasure {
             return (0, Some(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers")));
         }
 
+        // block_size/data_shards come from on-disk metadata; a corrupt FileInfo with a
+        // zero here must surface as an error, not a divide-by-zero panic on every GET.
+        if self.block_size == 0 || self.data_shards == 0 {
+            record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid);
+            return (0, Some(io::Error::new(ErrorKind::InvalidInput, "Invalid erasure coding parameters")));
+        }
+
         let Some(end_offset) = offset.checked_add(length) else {
             record_get_object_pipeline_failure(GET_STAGE_RANGE, GetObjectFailureReason::RangeOrLengthInvalid);
             return (0, Some(io::Error::new(ErrorKind::InvalidInput, "offset + length exceeds total length")));

crates/ecstore/src/erasure/coding/heal.rs

@@ -143,8 +143,11 @@ impl super::Erasure {
             let shard_size = self.shard_size().min(shard_file_size.saturating_sub(shard_offset));
             let (mut shards, errs) = read_heal_shards(&mut readers, shard_size, read_timeout).await;
 
-            // Data reads may use the first read quorum, but heal writes must only
-            // proceed when the source set is strong enough to validate itself.
+            // Every source shard is already bitrot-verified by its reader, so any
+            // data_shards survivors are sufficient to reconstruct — requiring more
+            // would make objects unhealable after losing exactly parity_shards disks,
+            // the failure EC is sized to tolerate. The parity cross-checks below stay
+            // opportunistic: they run whenever surplus source shards exist.
             let available_shards = errs.iter().filter(|e| e.is_none()).count();
             if available_shards < self.data_shards {
                 warn!(
@@ -157,19 +160,6 @@ impl super::Erasure {
                 return Err(Error::ErasureReadQuorum);
             }
 
-            let missing_data_source = shards.iter().take(self.data_shards).any(|shard| shard.is_none());
-            let required_shards = if missing_data_source && self.parity_shards > 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()

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 {

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<usize> = (part_index..=last_part_index).collect();
         debug!(bucket, object, ?part_indices, "Multipart part indices to stream");