Merge branch 'fix/get-dangling-latest-version' of https://github.com/rustfs/rustfs into fix/dangling-latest-read-before-write

Review of the dangling-latest metadata updates found three gaps:

- A version-only CopyObject wrote the new version and reverted on a
  write-quorum miss; refuse it before writing when the served version's
  disks cannot reach its write quorum, like update_object_meta_with_opts.
- Replication status write-back, including the zero-write ValidateOnly
  mode, now keeps its read-quorum error and never walks past dangling
  versions, since the walk queues heal for the versions it skips.
- The erasure-coding doc overstated that these callers never report a
  dangling key as absent; put_object_metadata and CopyObject keep their
  existing dangling cleanup when the walk finds no readable version.

.config/ecstore-required-tests.json

@@ -229,7 +229,7 @@
     {
       "invariant": "list-objects-shared-corpus-orphan-cleanup-requires-authoritative-evidence",
       "suite": "rustfs-ecstore",
-      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_orphan_cleanup_requires_authoritative_evidence"
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_orphan_listing_preserves_inputs"
     },
     {
       "invariant": "list-objects-shared-corpus-consumer-drop-joins-real-disk-producer",

crates/ecstore/src/disk/local.rs

@@ -10009,9 +10009,7 @@ impl LocalDisk {
                             } {
                                 meta.name.push_str(SLASH_SEPARATOR);
                                 // Conservative listings verify physical prefixes. Never-versioned
-                                // buckets use the bounded fast path, which only has to rule out
-                                // the data dirs a deleted version leaves behind; an empty listing
-                                // of such a prefix then reclaims committed residue.
+                                // buckets first rule out the data dirs a deleted version leaves behind.
                                 let listable = if opts.recursive || opts.incl_deleted {
                                     true
                                 } else if opts.skip_hidden_prefix_check {
@@ -10165,17 +10163,19 @@ impl LocalDisk {
     /// Reads stay small on genuine prefixes: each directory is sampled with one
     /// bounded batch, the sampled children are probed depth-first, and the
     /// directory is only read in full once every sampled child proved to hold
-    /// nothing listable. A total read budget makes a residue tree larger than
-    /// the budget surface and be hidden one level down instead, so the cost on
-    /// a genuine prefix never exceeds its depth in small directory reads. A
-    /// directory that vanishes mid-probe holds nothing listable.
+    /// nothing listable. If the read budget is exhausted, verify the subtree
+    /// before exposing the prefix. A directory that vanishes mid-probe holds
+    /// nothing listable.
     async fn directory_is_delete_residue(&self, bucket: &str, dir_name: &str, stall: Option<Duration>) -> Result<bool> {
         let dir_name = dir_name.trim_end_matches(SLASH_SEPARATOR);
         let mut reads = 0usize;
         let mut pending = vec![ResidueProbeStep::Sample(dir_name.to_owned())];
         while let Some(step) = pending.pop() {
             if reads >= DELETE_RESIDUE_PROBE_READ_BUDGET {
-                return Ok(false);
+                return self
+                    .directory_has_listing_entry(bucket, dir_name, false, stall)
+                    .await
+                    .map(|listable| !listable);
             }
             reads += 1;
 
@@ -11306,57 +11306,34 @@ impl LocalDisk {
 #[cfg(unix)]
 pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, usize)>) -> Vec<Result<Bytes>> {
     let n = requests.len();
-    let metrics = rustfs_io_metrics::LocalBatchReadMetrics::capture();
-    let submitted = metrics.as_ref().map(|_| std::time::Instant::now());
     tokio::task::spawn_blocking(move || {
-        use rustfs_io_metrics::LocalBatchReadStage;
         use std::os::unix::fs::FileExt;
 
-        let execution_started = metrics.as_ref().map(|_| std::time::Instant::now());
-        if let (Some(metrics), Some(submitted)) = (&metrics, submitted) {
-            metrics.record_stage(LocalBatchReadStage::QueueWait, true, submitted.elapsed());
-            metrics.record_batch(n);
-        }
         let mut results = Vec::with_capacity(n);
         for (file_path, offset, length) in requests {
-            let mut bytes_read = 0;
             let r = (|| -> Result<Bytes> {
-                let meta = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Metadata, || {
-                    std::fs::metadata(&file_path).map_err(DiskError::from)
-                })?;
+                let meta = std::fs::metadata(&file_path).map_err(DiskError::from)?;
                 let end = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?;
                 if meta.len() < u64::try_from(end).unwrap_or(u64::MAX) {
                     return Err(DiskError::FileCorrupt);
                 }
 
-                let file = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Open, || {
-                    std::fs::File::open(&file_path).map_err(DiskError::from)
-                })?;
+                let file = std::fs::File::open(&file_path).map_err(DiskError::from)?;
                 let mut buf = vec![0u8; length];
-                if length > 0 {
-                    measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Read, || {
-                        while bytes_read < length {
-                            let nbytes = file
-                                .read_at(&mut buf[bytes_read..], u64::try_from(offset + bytes_read).unwrap_or(u64::MAX))
-                                .map_err(DiskError::from)?;
-                            if nbytes == 0 {
-                                return Err(DiskError::FileCorrupt);
-                            }
-                            bytes_read += nbytes;
-                        }
-                        Ok(())
-                    })?;
+                let mut total = 0usize;
+                while total < length {
+                    let nbytes = file
+                        .read_at(&mut buf[total..], u64::try_from(offset + total).unwrap_or(u64::MAX))
+                        .map_err(DiskError::from)?;
+                    if nbytes == 0 {
+                        return Err(DiskError::FileCorrupt);
+                    }
+                    total += nbytes;
                 }
                 Ok(Bytes::from(buf))
             })();
-            if let Some(metrics) = &metrics {
-                metrics.record_shard(r.is_ok(), length, bytes_read);
-            }
             results.push(r);
         }
-        if let (Some(metrics), Some(started)) = (&metrics, execution_started) {
-            metrics.record_stage(LocalBatchReadStage::Execution, results.iter().all(Result::is_ok), started.elapsed());
-        }
         results
     })
     .await
@@ -11366,20 +11343,6 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize,
     })
 }
 
-#[cfg(unix)]
-fn measure_local_batch_stage<T>(
-    metrics: Option<&rustfs_io_metrics::LocalBatchReadMetrics>,
-    stage: rustfs_io_metrics::LocalBatchReadStage,
-    operation: impl FnOnce() -> Result<T>,
-) -> Result<T> {
-    let started = metrics.map(|_| std::time::Instant::now());
-    let result = operation();
-    if let (Some(metrics), Some(started)) = (metrics, started) {
-        metrics.record_stage(stage, result.is_ok(), started.elapsed());
-    }
-    result
-}
-
 #[async_trait::async_trait]
 impl DiskAPI for LocalDisk {
     async fn supports_guarded_delete_versions(&self) -> Result<()> {
@@ -25282,11 +25245,13 @@ mod test {
         )
         .await;
 
-        // A prefix with more residue directories than the probe budget stays
-        // visible rather than costing an unbounded walk.
+        // A prefix larger than the fast-probe budget still needs a complete
+        // read-only check before it can appear in a delimiter listing.
         for hour in 0..(DELETE_RESIDUE_PROBE_READ_BUDGET + 1) {
             write_residue(&bucket_dir.join(format!("bulk/2026/08/28/{hour:02}/a.parquet")), false).await;
+            write_residue(&bucket_dir.join(format!("bulk-live/2026/08/28/{hour:02}/a.parquet")), false).await;
         }
+        write_object(&bucket_dir.join("bulk-live/live.parquet"), "bulk-live/live.parquet").await;
         // Exactly at the budget: `edge/` with N object dirs costs one sampled
         // read of `edge`, two reads per object dir (its own and its data dir),
         // and one remainder read of `edge` once its 8-entry sample was full.
@@ -25367,10 +25332,8 @@ mod test {
             vec!["logs/default/2026/08/28/23/".to_owned()]
         );
         assert_eq!(scan_names(&disk, bucket, "logs/").await, vec!["logs/default/".to_owned()]);
-        assert_eq!(scan_names(&disk, bucket, "bulk/2026/08/").await, vec!["bulk/2026/08/28/".to_owned()]);
-        // Listed directly, every object dir is probed on its own and hidden;
-        // the budget only bites where the whole tree hangs off one entry, so
-        // the root listing below shows `edge-show/` but not `edge-hide/`.
+        assert_eq!(scan_names(&disk, bucket, "bulk/2026/08/").await, Vec::<String>::new());
+        assert_eq!(scan_names(&disk, bucket, "bulk-live/").await, vec!["bulk-live/live.parquet".to_owned()]);
         assert_eq!(scan_names(&disk, bucket, "edge-hide/").await, Vec::<String>::new());
         assert_eq!(scan_names(&disk, bucket, "edge-show/").await, Vec::<String>::new());
         assert_eq!(scan_names(&disk, bucket, "stray/").await, vec!["stray/2026/".to_owned()]);
@@ -25378,8 +25341,7 @@ mod test {
         assert_eq!(
             scan_names(&disk, bucket, "").await,
             vec![
-                "bulk/".to_owned(),
-                "edge-show/".to_owned(),
+                "bulk-live/".to_owned(),
                 "logs/".to_owned(),
                 "odd/".to_owned(),
                 "stray/".to_owned(),
@@ -30899,109 +30861,6 @@ mod test {
         );
     }
 
-    #[cfg(unix)]
-    #[test]
-    fn local_batch_read_metrics_cover_success_and_unattempted_failure_stages() {
-        use crate::test_metrics::CapturingRecorder;
-
-        let dir = tempfile::tempdir().expect("batch metrics directory");
-        let good = dir.path().join("good.bin");
-        std::fs::write(&good, b"good data").expect("write batch metrics payload");
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("batch metrics runtime");
-        let recorder = CapturingRecorder::default();
-        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
-        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
-        let results = metrics::with_local_recorder(&recorder, || {
-            runtime.block_on(batch_shard_pread(vec![
-                (good.clone(), 0, 9),
-                (dir.path().join("missing.bin"), 0, 4),
-                (good.clone(), 0, 1024),
-                (good, 0, 0),
-            ]))
-        });
-        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
-
-        assert_eq!(results[0].as_ref().expect("valid shard read").as_ref(), b"good data");
-        assert!(results[1].is_err(), "missing shard must remain an error");
-        assert!(matches!(results[2], Err(DiskError::FileCorrupt)), "range failure must preserve its error");
-        assert!(results[3].as_ref().expect("empty range succeeds").is_empty());
-        let counter = "rustfs_io_get_object_local_batch_shards_total";
-        assert_eq!(recorder.counter_value(counter, &[("outcome", "success")]), 2);
-        assert_eq!(recorder.counter_value(counter, &[("outcome", "error")]), 2);
-        assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 9);
-        let stages = "rustfs_io_get_object_local_batch_stage_seconds";
-        for (stage, success, errors) in [("metadata", 3, 1), ("open", 2, 0), ("read", 1, 0), ("execution", 0, 1)] {
-            assert_eq!(
-                recorder
-                    .histogram_values(stages, &[("stage", stage), ("outcome", "success")])
-                    .len(),
-                success
-            );
-            assert_eq!(
-                recorder
-                    .histogram_values(stages, &[("stage", stage), ("outcome", "error")])
-                    .len(),
-                errors
-            );
-        }
-        assert_eq!(recorder.histogram_values(stages, &[("stage", "queue_wait")]).len(), 1);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    fn local_batch_read_metrics_survive_caller_cancellation() {
-        use crate::test_metrics::CapturingRecorder;
-
-        let dir = tempfile::tempdir().expect("cancelled batch directory");
-        let path = dir.path().join("shard");
-        std::fs::write(&path, b"data").expect("write cancellation fixture");
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .max_blocking_threads(1)
-            .enable_all()
-            .build()
-            .expect("single blocking-worker runtime");
-        let recorder = CapturingRecorder::default();
-        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
-        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
-        metrics::with_local_recorder(&recorder, || {
-            runtime.block_on(async {
-                let (started_tx, started_rx) = tokio::sync::oneshot::channel();
-                let (release_tx, release_rx) = std::sync::mpsc::channel();
-                let blocker = tokio::task::spawn_blocking(move || {
-                    started_tx.send(()).expect("signal occupied blocking worker");
-                    release_rx.recv().expect("release occupied blocking worker");
-                });
-                started_rx.await.expect("blocking worker has started");
-                let mut batch = Box::pin(batch_shard_pread(vec![(path, 0, 4)]));
-                let pending = futures::FutureExt::now_or_never(&mut batch).is_none();
-                drop(batch);
-                rustfs_io_metrics::set_get_stage_metrics_enabled(false);
-                release_tx.send(()).expect("release worker after dropping caller");
-                blocker.await.expect("blocking worker finishes");
-                assert!(pending, "batch must have been submitted while the worker was occupied");
-                tokio::time::timeout(Duration::from_secs(5), async {
-                    while recorder
-                        .histogram_values("rustfs_io_get_object_local_batch_stage_seconds", &[("stage", "execution")])
-                        .is_empty()
-                    {
-                        tokio::task::yield_now().await;
-                    }
-                })
-                .await
-                .expect("detached batch must still record completion");
-            });
-        });
-        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
-        assert_eq!(
-            recorder.counter_value("rustfs_io_get_object_local_batch_shards_total", &[("outcome", "success")]),
-            1
-        );
-        assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 4);
-    }
-
     #[cfg(unix)]
     #[tokio::test]
     async fn test_batch_shard_pread_basic() {

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -1234,23 +1234,16 @@ async fn try_create_bitrot_readers_via_batch_pread(
     skip_verify_bitrot: bool,
 ) -> Option<BitrotReaderSetup> {
     use crate::disk::local::batch_shard_pread;
-    use rustfs_io_metrics::{LocalBatchReadDecision, record_local_batch_read_decision};
     use std::io::Cursor;
 
-    let prepare_started = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
-    let record_decision = |decision| {
-        record_local_batch_read_decision(decision, prepare_started.map(|started| started.elapsed()));
-    };
     let (adj_off, adj_len) = adjust_shard_read_params(read_offset, read_length, shard_size, &checksum_algo);
     if adj_len > object_mmap_read_max_length() {
-        record_decision(LocalBatchReadDecision::ReadCapExceeded);
         return None;
     }
 
     let mut batch_items: Vec<(usize, std::path::PathBuf, usize, usize)> = Vec::new();
     for (idx, disk_op) in disks.iter().enumerate() {
         if files.get(idx).is_some_and(|fi| fi.data.is_some()) {
-            record_decision(LocalBatchReadDecision::InlineData);
             return None;
         }
         if let Some(disk) = disk_op.as_ref() {
@@ -1258,20 +1251,12 @@ async fn try_create_bitrot_readers_via_batch_pread(
             let path_str = format!("{object}/{data_dir}/part.{part_number}");
             match disk.get_object_path_for_io_if_local(bucket, &path_str) {
                 Some(Ok(p)) => batch_items.push((idx, p, adj_off, adj_len)),
-                Some(Err(_)) => {
-                    record_decision(LocalBatchReadDecision::InvalidPath);
-                    return None;
-                }
-                None => {
-                    record_decision(LocalBatchReadDecision::NonLocalShard);
-                    return None;
-                }
+                _ => return None,
             }
         }
     }
 
     if batch_items.is_empty() {
-        record_decision(LocalBatchReadDecision::NoLocalShards);
         return None;
     }
 
@@ -1280,7 +1265,6 @@ async fn try_create_bitrot_readers_via_batch_pread(
     for (disk_index, _, _, _) in &batch_items {
         disk_call_counters::record(object, disk_call_counters::KIND_DISK_SHARD_OPEN, *disk_index);
     }
-    record_decision(LocalBatchReadDecision::Selected);
     let batch_results = batch_shard_pread(requests).await;
 
     let mut setup = BitrotReaderSetup::new(disks.len());
@@ -1329,7 +1313,6 @@ async fn try_create_bitrot_readers_via_batch_pread(
     _checksum_algo: HashAlgorithm,
     _skip_verify_bitrot: bool,
 ) -> Option<BitrotReaderSetup> {
-    rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::UnsupportedPlatform, None);
     None
 }
 
@@ -1493,19 +1476,11 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_with_prefere
 ) -> BitrotReaderSetup {
     let strategy = get_bitrot_reader_setup_strategy(mode, prefer_data_blocks_first);
 
-    let batch_read_enabled = if !use_mmap_read {
-        rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::MmapDisabled, None);
-        false
-    } else if matches!(
-        crate::set_disk::get_object_read_policy(),
-        crate::set_disk::GetObjectReadPolicy::CopySource
-    ) {
-        rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::CopySource, None);
-        false
-    } else {
-        true
-    };
-    if batch_read_enabled
+    if use_mmap_read
+        && !matches!(
+            crate::set_disk::get_object_read_policy(),
+            crate::set_disk::GetObjectReadPolicy::CopySource
+        )
         && let Some(mut setup) = try_create_bitrot_readers_via_batch_pread(
             files,
             disks,
@@ -5777,6 +5752,19 @@ impl SetDisks {
             return Ok(());
         }
 
+        // A metadata update that misses write quorum is not rolled back, so
+        // refuse it before writing when the target disks cannot reach quorum
+        // even if every write succeeds; otherwise the change a caller reports
+        // as failed would be served and spread by heal.
+        let write_quorum = fi.write_quorum(self.default_write_quorum());
+        let reachable = disks
+            .iter()
+            .map(|disk| disk.is_none().then_some(DiskError::DiskNotFound))
+            .collect::<Vec<_>>();
+        if let Some(err) = reduce_write_quorum_errs(&reachable, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+            return Err(err);
+        }
+
         self.invalidate_get_object_metadata_cache(bucket, object).await;
 
         let mut futures = Vec::with_capacity(disks.len());
@@ -5806,7 +5794,7 @@ impl SetDisks {
             }
         }
 
-        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, fi.write_quorum(self.default_write_quorum())) {
+        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
             return Err(err);
         }
 
@@ -6363,40 +6351,6 @@ impl SetDisks {
         Ok(purged)
     }
 
-    /// Reclaim metadata-less directory trees discovered by a recursive empty
-    /// bucket listing. The initial directory read fails closed if any slot is
-    /// unavailable; per-prefix purges then reuse the full cross-disk scan.
-    pub(crate) async fn purge_orphan_dir_objects_in_bucket(&self, bucket: &str) -> bool {
-        let disks = self.get_disks_internal().await;
-        if disks.is_empty() || disks.iter().any(Option::is_none) {
-            return false;
-        }
-
-        let mut prefixes = HashSet::new();
-        for disk in disks.iter().flatten() {
-            let entries = match disk.list_dir("", bucket, "", 0).await {
-                Ok(entries) => entries,
-                Err(_) => return false,
-            };
-            prefixes.extend(
-                entries
-                    .into_iter()
-                    .filter(|entry| entry.ends_with(SLASH_SEPARATOR) && is_safe_orphan_dir_entry(entry)),
-            );
-        }
-
-        let mut prefixes = prefixes.into_iter().collect::<Vec<_>>();
-        prefixes.sort_unstable();
-        let mut purged = false;
-        for prefix in prefixes {
-            match self.purge_orphan_dir_object(bucket, &prefix).await {
-                Ok(prefix_purged) => purged |= prefix_purged,
-                Err(_) => return false,
-            }
-        }
-        purged
-    }
-
     /// Orphan cleanup must not infer a complete scan from only the online
     /// subset of a set's disks.
     pub(crate) async fn orphan_purge_has_complete_disk_set(&self) -> bool {
@@ -7910,228 +7864,6 @@ mod tests {
         (dirs, disks)
     }
 
-    #[cfg(unix)]
-    #[test]
-    fn local_batch_read_decisions_follow_the_actual_reader_setup_entry() {
-        use crate::test_metrics::CapturingRecorder;
-
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("reader setup validation runtime");
-        let recorder = CapturingRecorder::default();
-        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
-        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
-        metrics::with_local_recorder(&recorder, || {
-            runtime.block_on(temp_env::async_with_vars(
-                [
-                    (rustfs_config::ENV_OBJECT_MMAP_READ_MAX_LENGTH, Some("4")),
-                    (rustfs_config::ENV_INTERNODE_PREWARM, Some("false")),
-                ],
-                async {
-                    let bucket = "batch-path-attribution";
-                    let object = "object";
-                    let (_dirs, disks) = call_counter_local_disks(bucket, 2).await;
-                    let data_dir = Uuid::new_v4();
-                    let mut files = vec![FileInfo::new(object, 1, 1); 2];
-                    for (file, disk) in files.iter_mut().zip(&disks) {
-                        file.data_dir = Some(data_dir);
-                        disk.as_ref()
-                            .expect("local disk")
-                            .write_all(bucket, &format!("{object}/{data_dir}/part.1"), Bytes::from_static(b"data"))
-                            .await
-                            .expect("write batch fixture");
-                    }
-                    assert_eq!(object_mmap_read_max_length(), 4, "cap must be controlled by this test");
-                    for length in [3, 4] {
-                        let setup = create_bitrot_readers_until_quorum_with_preference(
-                            &files,
-                            &disks,
-                            bucket,
-                            object,
-                            1,
-                            0,
-                            length,
-                            4,
-                            HashAlgorithm::None,
-                            false,
-                            true,
-                            1,
-                            1,
-                            BitrotReaderSetupMode::ReadQuorum,
-                            false,
-                            None,
-                            None,
-                        )
-                        .await;
-                        assert_eq!(setup.readers.iter().flatten().count(), 2);
-                    }
-                    assert!(
-                        try_create_bitrot_readers_via_batch_pread(
-                            &files,
-                            &disks,
-                            bucket,
-                            object,
-                            1,
-                            0,
-                            5,
-                            4,
-                            HashAlgorithm::None,
-                            false,
-                        )
-                        .await
-                        .is_none(),
-                        "a read above the adjusted-length cap must fall back"
-                    );
-                    assert!(
-                        try_create_bitrot_readers_via_batch_pread(
-                            &files,
-                            &disks,
-                            bucket,
-                            object,
-                            1,
-                            0,
-                            4,
-                            4,
-                            HashAlgorithm::HighwayHash256S,
-                            false,
-                        )
-                        .await
-                        .is_none(),
-                        "checksum overhead must be included in the read cap"
-                    );
-                    let mut inline_files = files.clone();
-                    inline_files[0].data = Some(Bytes::from_static(b"data"));
-                    assert!(
-                        try_create_bitrot_readers_via_batch_pread(
-                            &inline_files,
-                            &disks,
-                            bucket,
-                            object,
-                            1,
-                            0,
-                            4,
-                            4,
-                            HashAlgorithm::None,
-                            false,
-                        )
-                        .await
-                        .is_none()
-                    );
-                    let mut endpoint = Endpoint::try_from("http://127.0.0.1:1/disk").expect("remote placeholder endpoint");
-                    endpoint.is_local = false;
-                    let remote = new_disk(
-                        &endpoint,
-                        &DiskOption {
-                            cleanup: false,
-                            health_check: false,
-                        },
-                    )
-                    .await
-                    .expect("remote placeholder must initialize without a connection");
-                    assert!(
-                        try_create_bitrot_readers_via_batch_pread(
-                            &files,
-                            &[Some(remote), None],
-                            bucket,
-                            object,
-                            1,
-                            0,
-                            4,
-                            4,
-                            HashAlgorithm::None,
-                            false,
-                        )
-                        .await
-                        .is_none(),
-                        "any remote shard must keep the setup out of the local batch path"
-                    );
-                    assert!(
-                        try_create_bitrot_readers_via_batch_pread(
-                            &files,
-                            &disks,
-                            bucket,
-                            "../outside",
-                            1,
-                            0,
-                            4,
-                            4,
-                            HashAlgorithm::None,
-                            false,
-                        )
-                        .await
-                        .is_none()
-                    );
-                    let missing = vec![None, None];
-                    assert!(
-                        try_create_bitrot_readers_via_batch_pread(
-                            &files,
-                            &missing,
-                            bucket,
-                            object,
-                            1,
-                            0,
-                            4,
-                            4,
-                            HashAlgorithm::None,
-                            false,
-                        )
-                        .await
-                        .is_none()
-                    );
-                    for (use_mmap, policy) in [
-                        (false, crate::set_disk::GetObjectReadPolicy::Default),
-                        (true, crate::set_disk::GetObjectReadPolicy::CopySource),
-                    ] {
-                        crate::set_disk::with_get_object_read_policy(policy, async {
-                            create_bitrot_readers_until_quorum_with_preference(
-                                &files,
-                                &missing,
-                                bucket,
-                                object,
-                                1,
-                                0,
-                                4,
-                                4,
-                                HashAlgorithm::None,
-                                false,
-                                use_mmap,
-                                1,
-                                1,
-                                BitrotReaderSetupMode::ReadQuorum,
-                                false,
-                                None,
-                                None,
-                            )
-                            .await;
-                        })
-                        .await;
-                    }
-                },
-            ));
-        });
-        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
-        let metric = "rustfs_io_get_object_local_batch_decisions_total";
-        for (decision, expected) in [
-            ("selected", 2),
-            ("read_cap_exceeded", 2),
-            ("inline_data", 1),
-            ("invalid_path", 1),
-            ("non_local_shard", 1),
-            ("no_local_shards", 1),
-            ("mmap_disabled", 1),
-            ("copy_source", 1),
-        ] {
-            assert_eq!(recorder.counter_value(metric, &[("decision", decision)]), expected, "{decision}");
-        }
-        assert_eq!(
-            recorder.counter_value("rustfs_io_get_object_local_batch_shards_total", &[("outcome", "success")]),
-            4
-        );
-        assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 14);
-        assert_eq!(recorder.histogram_sample_count("rustfs_io_get_object_local_batch_prepare_seconds"), 8);
-    }
-
     async fn reopen_local_disk(dir: &TempDir) -> DiskStore {
         let endpoint =
             Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
@@ -9721,7 +9453,7 @@ mod tests {
             "lifecycle-inline-data-get-fanout-bucket",
             "lifecycle-inline-data-get-object",
             GET_OBJECT_PATH_LEGACY_DUPLEX,
-            None,
+            false,
         );
     }
 
@@ -9732,7 +9464,7 @@ mod tests {
             "lifecycle-bounded-inline-data-get-fanout-bucket",
             "lifecycle-bounded-inline-data-get-object",
             GET_OBJECT_PATH_LEGACY_DUPLEX,
-            Some("true"),
+            true,
         );
     }
 
@@ -9743,7 +9475,7 @@ mod tests {
             RUSTFS_META_BUCKET,
             "buckets/.usage-cache/lifecycle-inline-data-get-object",
             GET_OBJECT_PATH_INTERNAL_META,
-            None,
+            false,
         );
     }
 
@@ -9844,7 +9576,7 @@ mod tests {
         bucket: &'static str,
         object_prefix: &str,
         expected_path: &'static str,
-        bounded_fanout_env: Option<&'static str>,
+        bounded_fanout: bool,
     ) {
         const DISKS: usize = 4;
         let object = object_with_initial_data_shards(bucket, object_prefix, 2, 3);
@@ -9864,16 +9596,26 @@ mod tests {
                     [
                         ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
                         ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
-                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", bounded_fanout_env),
+                        (
+                            "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT",
+                            Some(if bounded_fanout { "true" } else { "false" }),
+                        ),
                     ],
                     async {
-                        let barrier_disk = if bounded_fanout_env.is_some() {
-                            bounded_spare_disk_index(bucket, &object, 2, 2)
-                        } else {
-                            3
-                        };
+                        let barrier_disk = bounded_spare_disk_index(bucket, &object, 2, 2);
                         let barrier =
                             rename_fanout_barrier::arm(&object, barrier_disk, rename_fanout_barrier::PHASE_READ_VERSION);
+                        // Keep one initial bounded read pending so its hedge is scheduled;
+                        // eager fanout needs all quorum reads parked until the spare starts.
+                        let quorum_disks = if bounded_fanout {
+                            vec![bounded_metadata_fanout_order(bucket, &object, DISKS, 2)[0]]
+                        } else {
+                            (0..DISKS).filter(|&index| index != barrier_disk).collect()
+                        };
+                        let quorum_barriers = quorum_disks
+                            .into_iter()
+                            .map(|index| rename_fanout_barrier::arm(&object, index, rename_fanout_barrier::PHASE_READ_VERSION))
+                            .collect::<Vec<_>>();
                         let disks_for_read = disks.clone();
                         let object_for_read = object.clone();
                         let mut read = tokio::spawn(async move {
@@ -9895,6 +9637,9 @@ mod tests {
                         tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
                             .await
                             .expect("metadata fanout should pause the spare metadata task");
+                        for quorum_barrier in &quorum_barriers {
+                            quorum_barrier.release();
+                        }
                         let (parts_metadata, errs, diagnostics) = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read)
                             .await
                             .expect("verified inline quorum should abort the paused eager fanout")

crates/ecstore/src/set_disk/core/metadata_quorum.rs

@@ -597,10 +597,8 @@ mod tests {
                         .iter()
                         .map(|slot| slot.file_info().expect("every disk must have arrived").clone())
                         .collect::<Vec<_>>();
-                    let (_, selected, selection_quorum) =
-                        SetDisks::select_valid_fileinfo(&vec![None; 4], &completed_metadata, &vec![None; 4], "", 2, 3)
-                            .expect("three matching B disks must select the current identity");
-                    assert_eq!(selection_quorum, 3);
+                    let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; 4], &completed_metadata, &vec![None; 4], 2)
+                        .expect("three matching B disks must select the current identity");
                     assert_eq!(selected.version_id, expected_current.version_id);
                     assert_eq!(selected.mod_time, expected_current.mod_time);
                     assert_eq!(selected.data_dir, expected_current.data_dir);

crates/ecstore/src/set_disk/metadata.rs

@@ -665,14 +665,20 @@ impl SetDisks {
         read_quorum
     }
 
-    pub(super) fn select_valid_fileinfo(
+    /// The commit-proof rule kept for internal metadata in the system bucket.
+    /// Pool metadata's PREPARE/COMMIT recovery and identity epochs treat an
+    /// ordinary read as proof that a record committed: when enough copies
+    /// answer to prove write quorum, the latest version must hold it, and a
+    /// degraded read never serves a newer version below write quorum. A split
+    /// write therefore stays unreadable until native recovery resolves it.
+    fn select_committed_fileinfo(
         disks: &[Option<DiskStore>],
         parts_metadata: &[FileInfo],
         errs: &[Option<DiskError>],
         version_id: &str,
         read_quorum: usize,
         write_quorum: usize,
-    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo, usize)> {
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo)> {
         let selection_quorum =
             Self::latest_fileinfo_selection_quorum(version_id, parts_metadata, errs, read_quorum, write_quorum);
         let (usable_metadata, has_read_error) = Self::usable_fileinfo_count(parts_metadata, errs);
@@ -683,14 +689,71 @@ impl SetDisks {
             && usable_metadata >= write_quorum
             && selection_quorum == read_quorum
         {
-            let (online_disks, fi) = Self::pick_degraded_latest_fileinfo(disks, parts_metadata, errs, read_quorum, write_quorum)?;
-            return Ok((online_disks, fi, read_quorum));
+            return Self::pick_degraded_latest_fileinfo(disks, parts_metadata, errs, read_quorum, write_quorum);
         }
 
         let (online_disks, mod_time, etag) = Self::list_online_disks(disks, parts_metadata, errs, selection_quorum);
         let fi = Self::pick_valid_fileinfo(parts_metadata, mod_time, etag, selection_quorum)?;
 
-        Ok((online_disks, fi, selection_quorum))
+        Ok((online_disks, fi))
+    }
+
+    /// Select the version a read of `bucket` serves: S3 buckets use MinIO's
+    /// read-quorum rule (`select_valid_fileinfo`); the system bucket keeps the
+    /// commit-proof rule (`select_committed_fileinfo`).
+    #[allow(clippy::too_many_arguments)]
+    pub(super) fn select_fileinfo_for_read(
+        bucket: &str,
+        disks: &[Option<DiskStore>],
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        version_id: &str,
+        read_quorum: usize,
+        write_quorum: usize,
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo)> {
+        if crate::bucket::utils::is_meta_bucketname(bucket) {
+            Self::select_committed_fileinfo(disks, parts_metadata, errs, version_id, read_quorum, write_quorum)
+        } else {
+            Self::select_valid_fileinfo(disks, parts_metadata, errs, read_quorum)
+        }
+    }
+
+    /// Select the version an S3 read serves with MinIO's rule (`getObjectFileInfo`
+    /// -> `listOnlineDisks` -> `pickValidFileInfo`): the modification time
+    /// shared by the most copies wins when it reaches `read_quorum`, a tie
+    /// goes to the newer time, and one content identity at that time must
+    /// itself reach `read_quorum`. Heal picks the latest version the same
+    /// way, so a read returns the version heal converges the set to; a write
+    /// that failed after reaching exactly `read_quorum` copies is readable
+    /// instead of failing until heal runs, as listing already shows it.
+    ///
+    /// `read_quorum` follows the parity most copies agree on, which delete
+    /// markers and empty versions can lower below the selected version's own
+    /// data blocks; like MinIO's `getLatestFileInfo`, a version with local
+    /// shards also needs that many copies, or HEAD would report a version
+    /// GET cannot decode.
+    pub(super) fn select_valid_fileinfo(
+        disks: &[Option<DiskStore>],
+        parts_metadata: &[FileInfo],
+        errs: &[Option<DiskError>],
+        read_quorum: usize,
+    ) -> disk::error::Result<(Vec<Option<DiskStore>>, FileInfo)> {
+        let (online_disks, mod_time, etag) = Self::list_online_disks(disks, parts_metadata, errs, read_quorum);
+        let (fi, copies) = Self::find_file_info_in_quorum_counted(parts_metadata, &mod_time, &etag, read_quorum)?;
+        if !fi.deleted && !fi.is_remote() && fi.size > 0 && copies < fi.erasure.data_blocks {
+            // The shortcut counts byte-identical copies only; count every copy
+            // of the content identity before refusing the version.
+            let identity = Self::file_info_quorum_hash(&fi);
+            let copies = parts_metadata
+                .iter()
+                .filter(|meta| file_info_is_valid_for_metadata(meta) && Self::file_info_quorum_hash(meta) == identity)
+                .count();
+            if copies < fi.erasure.data_blocks {
+                return Err(DiskError::ErasureReadQuorum);
+            }
+        }
+
+        Ok((online_disks, fi))
     }
 
     pub(super) fn pick_valid_fileinfo(
@@ -1025,6 +1088,23 @@ impl SetDisks {
         etag: &Option<String>,
         quorum: usize,
     ) -> disk::error::Result<FileInfo> {
+        Self::find_file_info_in_quorum_counted(metas, mod_time, etag, quorum).map(|(fi, _)| fi)
+    }
+
+    /// `find_file_info_in_quorum`, also returning how many copies carry the
+    /// selected content identity.
+    ///
+    /// Two identities can share the top count only in a data = parity set
+    /// split exactly in half at one modification time, such as a metadata
+    /// update that failed after reaching half the drives. The identity held
+    /// by the lowest-index disk wins, so every read and heal picks the same
+    /// side and a retried metadata write can repair the split.
+    fn find_file_info_in_quorum_counted(
+        metas: &[FileInfo],
+        mod_time: &Option<OffsetDateTime>,
+        etag: &Option<String>,
+        quorum: usize,
+    ) -> disk::error::Result<(FileInfo, usize)> {
         if quorum < 1 {
             warn!("find_file_info_in_quorum: quorum < 1");
             return Err(DiskError::ErasureReadQuorum);
@@ -1053,7 +1133,7 @@ impl SetDisks {
             if same_count >= quorum && same_count > total_valid - same_count {
                 let mut fi = first.clone();
                 fi.is_latest = fi.successor_mod_time.is_none();
-                return Ok(fi);
+                return Ok((fi, same_count));
             }
         }
 
@@ -1106,15 +1186,12 @@ impl SetDisks {
             *count_map.entry(hash).or_insert(0) += 1;
         }
 
-        let mut max_val = None;
-        let mut max_count = 0;
-
-        for (&val, &count) in &count_map {
-            if count > max_count {
-                max_val = Some(val);
-                max_count = count;
-            }
-        }
+        let max_count = count_map.values().copied().max().unwrap_or(0);
+        let max_val = meta_hashes
+            .iter()
+            .flatten()
+            .find(|hash| count_map.get(*hash) == Some(&max_count))
+            .copied();
 
         if max_count < quorum {
             warn!(
@@ -1155,17 +1232,24 @@ impl SetDisks {
         if found {
             let mut fi = found_fi.expect("operation should succeed");
 
-            for (val, &count) in &valid_obj_map {
-                if count >= quorum {
-                    fi.successor_mod_time = val.successor_mod_time;
-                    fi.num_versions = val.num_versions;
-                    fi.is_latest = val.successor_mod_time.is_none();
-
-                    break;
-                }
+            // Scan in disk order so that two version-stack shapes with equal
+            // counts resolve the same way on every read.
+            let quorum_props = metas
+                .iter()
+                .zip(meta_hashes.iter())
+                .filter(|(meta, hash)| hash.is_some() && **hash == max_val && file_info_is_valid_for_metadata(meta))
+                .map(|(meta, _)| ObjProps {
+                    successor_mod_time: meta.successor_mod_time,
+                    num_versions: meta.num_versions,
+                })
+                .find(|props| valid_obj_map.get(props).is_some_and(|&count| count >= quorum));
+            if let Some(props) = quorum_props {
+                fi.successor_mod_time = props.successor_mod_time;
+                fi.num_versions = props.num_versions;
+                fi.is_latest = props.successor_mod_time.is_none();
             }
 
-            return Ok(fi);
+            return Ok((fi, max_count));
         }
 
         warn!("find_file_info_in_quorum: fileinfo not found");

crates/ecstore/src/set_disk/mod.rs

@@ -1024,6 +1024,19 @@ impl GetObjectFileInfo {
         self.as_parts().0
     }
 
+    /// Marks an exact-version selection as the object's latest version after
+    /// the versions above it were skipped as dangling.
+    fn mark_selected_as_latest(&mut self) {
+        if let Some(OwnedGetObjectFileInfo {
+            payload: OwnedGetObjectFileInfoPayload::Direct { fi, .. },
+            ..
+        }) = self.owned.as_mut()
+        {
+            fi.is_latest = true;
+            fi.successor_mod_time = None;
+        }
+    }
+
     fn parts_metadata(&self) -> &[FileInfo] {
         self.as_parts().1
     }
@@ -1790,7 +1803,7 @@ mod prepared_get_object_metadata_tests {
         assert!(original.is_complete());
         let (original_metadata, original_errors, _) = original.into_legacy();
         assert!(original_errors.iter().all(Option::is_none));
-        let (_, generation_a, _) = SetDisks::select_valid_fileinfo(&disks, &original_metadata, &original_errors, "", 2, 3)
+        let (_, generation_a) = SetDisks::select_valid_fileinfo(&disks, &original_metadata, &original_errors, 2)
             .expect("generation A must have real metadata quorum");
         assert!(!generation_a.inline_data(), "fixture must use non-inline shard files");
         let order = bounded_metadata_fanout_order(bucket, &object, 4, 2);
@@ -1851,8 +1864,8 @@ mod prepared_get_object_metadata_tests {
                 assert!(replacement.is_complete());
                 let (replacement_metadata, replacement_errors, _) = replacement.into_legacy();
                 assert!(replacement_errors.iter().all(Option::is_none));
-                let (online, generation_b, _) =
-                    SetDisks::select_valid_fileinfo(&disks, &replacement_metadata, &replacement_errors, "", 2, 3)
+                let (online, generation_b) =
+                    SetDisks::select_valid_fileinfo(&disks, &replacement_metadata, &replacement_errors, 2)
                         .expect("replacement must have healthy real metadata quorum");
                 assert_eq!(online.iter().flatten().count(), 4);
                 assert_ne!(generation_a.data_dir, generation_b.data_dir);
@@ -6829,7 +6842,7 @@ impl SetDisks {
         let (metadata, errors) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false).await?;
         // Match delete_object_version_with_purge, independently of data parity or online inventory size.
         let quorum = self.set_drive_count / 2 + 1;
-        let (_, marker, _) = Self::select_valid_fileinfo(&disks, &metadata, &errors, version_id, quorum, quorum)
+        let (_, marker) = Self::select_valid_fileinfo(&disks, &metadata, &errors, quorum)
             .map_err(|error| to_object_err(error.into(), vec![bucket, object, version_id]))?;
         if !marker.deleted || marker.tier_free_version() || marker.version_id.unwrap_or_else(Uuid::nil) != requested_version {
             return Err(Error::DataMovementOverwriteErr(
@@ -10475,13 +10488,6 @@ mod tests {
 
         assert!(!purged, "an incomplete disk scan must not claim the tree is an orphan");
         assert!(prefix_dir.join("nested/leaf").exists(), "online disk contents must remain untouched");
-
-        let bucket_purged = set.purge_orphan_dir_objects_in_bucket("bucket").await;
-        assert!(!bucket_purged, "bucket-wide cleanup must fail closed with an unavailable disk slot");
-        assert!(
-            prefix_dir.join("nested/leaf").exists(),
-            "bucket-wide cleanup must preserve the online tree"
-        );
     }
 
     // Cross-disk safety: if any drive still holds object data under the prefix, refuse
@@ -11976,41 +11982,121 @@ mod tests {
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_quorum_requires_write_quorum_when_full_metadata_is_available() {
+    fn test_latest_fileinfo_selection_breaks_same_time_tie_in_disk_order() {
         let mod_time = OffsetDateTime::now_utc();
         let data_dir = Uuid::new_v4();
-        let metas = vec![
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 1),
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 2),
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-b", 3),
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-b", 4),
-        ];
+        let tagged = |index, tag: &str| {
+            let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+            fi.metadata.insert("x-amz-tagging".to_string(), tag.to_string());
+            fi
+        };
         let errs = vec![None, None, None, None];
 
-        let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3);
+        // A tagging write that failed after reaching half the drives.
+        for (metas, expected) in [
+            (vec![tagged(1, "a=1"), tagged(2, "a=1"), tagged(3, "a=2"), tagged(4, "a=2")], "a=1"),
+            (vec![tagged(1, "a=2"), tagged(2, "a=1"), tagged(3, "a=1"), tagged(4, "a=2")], "a=2"),
+        ] {
+            // Each call hashes into fresh maps; a map-order choice would flip.
+            for _ in 0..32 {
+                let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                    .expect("a same-time half split serves the side on the lowest-index disk");
+                let healed =
+                    SetDisks::find_file_info_in_quorum(&metas, &Some(mod_time), &None, 2).expect("heal resolves the same split");
 
-        assert_eq!(quorum, 3);
+                assert_eq!(selected.metadata.get("x-amz-tagging").map(String::as_str), Some(expected));
+                assert_eq!(healed.metadata.get("x-amz-tagging").map(String::as_str), Some(expected));
+            }
+        }
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_quorum_preserves_read_quorum_for_version_or_degraded_reads() {
-        let mod_time = OffsetDateTime::now_utc();
-        let data_dir = Uuid::new_v4();
-        let metas = vec![
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 1),
-            quorum_test_fileinfo(mod_time, data_dir, "part-etag-a", 2),
-            FileInfo::default(),
-            FileInfo::default(),
-        ];
-        let degraded_errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)];
-        let clean_errs = vec![None, None, None, None];
+    fn test_latest_fileinfo_selection_requires_the_versions_own_data_blocks() {
+        let old_mod_time = OffsetDateTime::now_utc();
+        let new_mod_time = old_mod_time + time::Duration::seconds(1);
+        let old = |index| quorum_test_fileinfo(old_mod_time, Uuid::from_u128(1), "part-etag-old", index);
+        let new = |index| {
+            let mut fi = quorum_test_fileinfo(new_mod_time, Uuid::from_u128(2), "part-etag-new", index);
+            fi.erasure.data_blocks = 3;
+            fi.erasure.parity_blocks = 1;
+            fi
+        };
+        let errs = vec![None, None, None, None];
 
-        assert_eq!(SetDisks::latest_fileinfo_selection_quorum("", &metas, &degraded_errs, 2, 3), 2);
-        assert_eq!(SetDisks::latest_fileinfo_selection_quorum("version-id", &metas, &clean_errs, 2, 3), 2);
+        // Read quorum two comes from the 2+2 copies; the newer 3+1 version
+        // wins the time tie but cannot be decoded from two shards.
+        let metas = vec![old(1), old(2), new(3), new(4)];
+        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2);
+        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
+
+        let metas = vec![old(1), new(2), new(3), new(4)];
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("three copies decode a 3+1 version");
+        assert_eq!(selected.mod_time, Some(new_mod_time));
+
+        // Replication-status drift does not change the content identity, so
+        // the drifted copy still counts toward the version's data blocks.
+        let replication_status_key = format!(
+            "{}{}",
+            rustfs_utils::http::RUSTFS_INTERNAL_PREFIX,
+            rustfs_utils::http::SUFFIX_REPLICATION_STATUS
+        );
+        let mut drifted = new(4);
+        drifted
+            .metadata
+            .insert(replication_status_key, "target-a=PENDING;".to_string());
+        let metas = vec![old(1), new(2), new(3), drifted];
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("a replication-status drift still leaves three copies of the version");
+        assert_eq!(selected.mod_time, Some(new_mod_time));
+    }
+
+    #[tokio::test]
+    async fn update_object_meta_below_write_quorum_writes_nothing() {
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+        use crate::storage_api_contracts::object::ObjectIO as _;
+
+        let (_dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks(4, 2).await;
+        let (bucket, object) = ("metadata-write-quorum", "object");
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        set.put_object(
+            bucket,
+            object,
+            &mut crate::object_api::PutObjReader::from_vec(b"body".to_vec()),
+            &ObjectOptions::default(),
+        )
+        .await
+        .expect("object should commit on every drive");
+        let disks = set.get_disks_internal().await;
+        let (metadata, _) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", false, false, false)
+            .await
+            .expect("committed metadata should be readable");
+        let mut update = metadata[0].clone();
+        update.metadata.insert("x-amz-meta-probe".to_string(), "written".to_string());
+
+        // A read that selected one side of a 2+2 split returns two disks.
+        let mut targets = disks.clone();
+        targets[2] = None;
+        targets[3] = None;
+        let err = set
+            .update_object_meta(bucket, object, update, &targets)
+            .await
+            .expect_err("two of four disks cannot reach write quorum");
+
+        assert_eq!(err, DiskError::ErasureWriteQuorum);
+        let (after, _) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", false, false, false)
+            .await
+            .expect("metadata should stay readable");
+        assert!(
+            after.iter().all(|fi| !fi.metadata.contains_key("x-amz-meta-probe")),
+            "a refused update must not reach any disk"
+        );
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_quorum_keeps_read_quorum_for_partial_overwrite_with_read_error() {
+    fn test_latest_fileinfo_selection_keeps_older_majority_over_partial_overwrite() {
         let old_mod_time = OffsetDateTime::now_utc();
         let new_mod_time = old_mod_time + time::Duration::seconds(1);
         let old_data_dir = Uuid::new_v4();
@@ -12023,26 +12109,15 @@ mod tests {
         ];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3);
-        let (online_disks, mod_time, etag) = SetDisks::list_online_disks(&vec![None; metas.len()], &metas, &errs, quorum);
-        let fi = SetDisks::pick_valid_fileinfo(&metas, mod_time, etag, quorum)
-            .expect("old metadata should remain readable with read quorum");
-
-        assert_eq!(quorum, 2);
-        assert_eq!(online_disks.len(), metas.len());
-        assert_eq!(fi.data_dir, Some(old_data_dir));
-        assert_eq!(fi.parts[0].etag, "part-etag-old");
-
-        let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
             .expect("old metadata should remain selectable with read quorum");
-        assert_eq!(selected_quorum, 2);
         assert_eq!(selected.data_dir, Some(old_data_dir));
         assert_eq!(selected.parts[0].etag, "part-etag-old");
         assert!(selected.is_latest);
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_rejects_partial_latest_read_quorum_with_read_error() {
+    fn test_latest_fileinfo_selection_serves_newer_read_quorum_with_read_error() {
         let old_mod_time = OffsetDateTime::now_utc();
         let new_mod_time = old_mod_time + time::Duration::seconds(1);
         let old_data_dir = Uuid::new_v4();
@@ -12055,9 +12130,120 @@ mod tests {
         ];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3);
+        // The new version may hold write quorum through the missing disk; the
+        // old one cannot, and heal converges the set to the new version.
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("the newer version reaches read quorum");
 
-        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
+        assert_eq!(selected.data_dir, Some(new_data_dir));
+        assert_eq!(selected.mod_time, Some(new_mod_time));
+    }
+
+    #[test]
+    fn test_latest_fileinfo_selection_serves_newer_side_of_complete_split() {
+        let old_mod_time = OffsetDateTime::now_utc();
+        let new_mod_time = old_mod_time + time::Duration::seconds(1);
+        let old_data_dir = Uuid::new_v4();
+        let new_data_dir = Uuid::new_v4();
+        let old = |index| quorum_test_fileinfo(old_mod_time, old_data_dir, "part-etag-old", index);
+        let new = |index| quorum_test_fileinfo(new_mod_time, new_data_dir, "part-etag-new", index);
+        let errs = vec![None, None, None, None];
+
+        for metas in [vec![old(1), old(2), new(3), new(4)], vec![new(1), old(2), new(3), old(4)]] {
+            let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                .expect("a 2+2 split with every disk answering serves the newer version");
+
+            assert_eq!(selected.data_dir, Some(new_data_dir));
+            assert_eq!(selected.parts[0].etag, "part-etag-new");
+            assert!(selected.is_latest);
+        }
+    }
+
+    #[test]
+    fn test_system_bucket_reads_keep_the_commit_proof_rule() {
+        let old_mod_time = OffsetDateTime::now_utc();
+        let new_mod_time = old_mod_time + time::Duration::seconds(1);
+        let old_data_dir = Uuid::new_v4();
+        let new_data_dir = Uuid::new_v4();
+        let old = |index| quorum_test_fileinfo(old_mod_time, old_data_dir, "part-etag-old", index);
+        let new = |index| quorum_test_fileinfo(new_mod_time, new_data_dir, "part-etag-new", index);
+
+        // A PREPARE/COMMIT split of pool metadata must stay unreadable so that
+        // native recovery, not an ordinary read, decides whether it committed.
+        for (metas, errs) in [
+            (vec![old(1), old(2), new(3), new(4)], vec![None, None, None, None]),
+            (
+                vec![new(1), new(2), old(3), FileInfo::default()],
+                vec![None, None, None, Some(DiskError::DiskNotFound)],
+            ),
+        ] {
+            let disks = vec![None; metas.len()];
+            let (_, served) = SetDisks::select_fileinfo_for_read("bucket", &disks, &metas, &errs, "", 2, 3)
+                .expect("an S3 read serves the newer side at read quorum");
+            assert_eq!(served.data_dir, Some(new_data_dir));
+
+            let internal = SetDisks::select_fileinfo_for_read(crate::disk::RUSTFS_META_BUCKET, &disks, &metas, &errs, "", 2, 3);
+            assert!(matches!(internal, Err(DiskError::ErasureReadQuorum)));
+        }
+
+        let committed = vec![new(1), new(2), new(3), old(4)];
+        let (_, internal) = SetDisks::select_fileinfo_for_read(
+            crate::disk::RUSTFS_META_BUCKET,
+            &vec![None; committed.len()],
+            &committed,
+            &[None, None, None, None],
+            "",
+            2,
+            3,
+        )
+        .expect("a write-quorum version stays readable in the system bucket");
+        assert_eq!(internal.data_dir, Some(new_data_dir));
+    }
+
+    #[test]
+    fn test_latest_fileinfo_selection_picks_version_stack_shape_in_disk_order() {
+        let mod_time = OffsetDateTime::now_utc();
+        let data_dir = Uuid::new_v4();
+        let successor = mod_time + time::Duration::seconds(1);
+        let shape = |index, successor_mod_time: Option<OffsetDateTime>, num_versions| {
+            let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+            fi.successor_mod_time = successor_mod_time;
+            fi.is_latest = successor_mod_time.is_none();
+            fi.num_versions = num_versions;
+            fi
+        };
+        let errs = vec![None, None, None, None];
+
+        for (metas, expected) in [
+            (
+                vec![
+                    shape(1, Some(successor), 2),
+                    shape(2, None, 1),
+                    shape(3, Some(successor), 2),
+                    shape(4, None, 1),
+                ],
+                Some(successor),
+            ),
+            (
+                vec![
+                    shape(1, None, 1),
+                    shape(2, Some(successor), 2),
+                    shape(3, None, 1),
+                    shape(4, Some(successor), 2),
+                ],
+                None,
+            ),
+        ] {
+            // Each call hashes into fresh maps; a map-order choice would flip.
+            for _ in 0..32 {
+                let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                    .expect("one content identity reaches read quorum");
+
+                assert_eq!(selected.successor_mod_time, expected);
+                assert_eq!(selected.is_latest, expected.is_none());
+                assert_eq!(selected.num_versions, if expected.is_none() { 1 } else { 2 });
+            }
+        }
     }
 
     #[test]
@@ -12075,11 +12261,9 @@ mod tests {
         let metas = vec![first, second, FileInfo::default(), FileInfo::default()];
         let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)];
 
-        let (_, mut selected, selected_quorum) =
-            SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
-                .expect("read quorum should remain enough when no competing latest is visible");
+        let (_, mut selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("read quorum should remain enough when no competing latest is visible");
 
-        assert_eq!(selected_quorum, 2);
         assert_eq!(selected.data_dir, Some(data_dir));
         assert_eq!(selected.parts[0].etag, "part-etag-old");
         assert!(selected.parts[0].checksums.is_none());
@@ -12106,7 +12290,7 @@ mod tests {
         let metas = vec![first, second, FileInfo::default(), FileInfo::default()];
         let errs = vec![None, None, Some(DiskError::DiskNotFound), Some(DiskError::DiskNotFound)];
 
-        let (_, mut selected, _) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
+        let (_, mut selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
             .expect("winner selection should defer sidecar decoding");
         let err = SetDisks::hydrate_selected_fileinfo_part_checksums(&mut selected)
             .expect_err("a malformed degraded winner must fail closed when checksums are requested");
@@ -12173,10 +12357,9 @@ mod tests {
         let metas = vec![latest_meta, stale_stack_meta, newer_stack_meta, FileInfo::default()];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
             .expect("same object version should stay readable despite derived version stack drift");
 
-        assert_eq!(selected_quorum, 3);
         assert_eq!(selected.data_dir, Some(data_dir));
         assert_eq!(selected.parts[0].etag, "part-etag");
         assert_eq!(selected.mod_time, Some(mod_time));
@@ -12269,81 +12452,77 @@ mod tests {
         let metas = vec![meta_a, meta_b, meta_c, FileInfo::default()];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let (_, selected, selected_quorum) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3)
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
             .expect("replication status drift should not split readable object identity");
 
-        assert_eq!(selected_quorum, 3);
         assert_eq!(selected.data_dir, Some(data_dir));
         assert_eq!(selected.parts[0].etag, "part-etag");
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_rejects_same_modtime_metadata_split_without_write_quorum() {
+    fn test_latest_fileinfo_selection_serves_same_time_metadata_majority() {
         let mod_time = OffsetDateTime::now_utc();
         let data_dir = Uuid::new_v4();
-        let mut old_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
-        let mut old_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2);
-        let mut partial_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3);
-        old_meta_a.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string());
-        old_meta_b.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string());
-        partial_meta
-            .metadata
-            .insert("x-amz-meta-color".to_string(), "red".to_string());
-        let metas = vec![old_meta_a, old_meta_b, partial_meta, FileInfo::default()];
+        let colored = |index, color: &str| {
+            let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+            fi.metadata.insert("x-amz-meta-color".to_string(), color.to_string());
+            fi
+        };
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3);
-        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3);
+        // With one disk missing, only the two-copy side can hold write quorum,
+        // whichever disk order the copies arrive in.
+        for (metas, expected) in [
+            (
+                vec![colored(1, "blue"), colored(2, "blue"), colored(3, "red"), FileInfo::default()],
+                "blue",
+            ),
+            (vec![colored(1, "blue"), colored(2, "red"), colored(3, "red"), FileInfo::default()], "red"),
+        ] {
+            let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                .expect("the read-quorum side of a same-time metadata split is served");
 
-        assert_eq!(quorum, 2);
-        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
+            assert_eq!(selected.metadata.get("x-amz-meta-color").map(String::as_str), Some(expected));
+        }
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_rejects_same_modtime_partial_metadata_read_quorum() {
+    fn test_latest_fileinfo_selection_serves_same_time_transition_majority() {
         let mod_time = OffsetDateTime::now_utc();
         let data_dir = Uuid::new_v4();
-        let mut old_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
-        let mut partial_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2);
-        let mut partial_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3);
-        old_meta.metadata.insert("x-amz-meta-color".to_string(), "blue".to_string());
-        partial_meta_a
-            .metadata
-            .insert("x-amz-meta-color".to_string(), "red".to_string());
-        partial_meta_b
-            .metadata
-            .insert("x-amz-meta-color".to_string(), "red".to_string());
-        let metas = vec![old_meta, partial_meta_a, partial_meta_b, FileInfo::default()];
+        let local = |index| quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+        let remote = |index| {
+            let mut fi = quorum_test_fileinfo(mod_time, data_dir, "part-etag", index);
+            fi.transition_status = TRANSITION_COMPLETE.to_string();
+            fi.transition_tier = "WARM".to_string();
+            fi.transitioned_objname = "remote/object".to_string();
+            fi.transition_version_id = Some(Uuid::from_u128(7));
+            fi
+        };
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3);
+        for (metas, expect_remote) in [
+            (vec![local(1), local(2), remote(3), FileInfo::default()], false),
+            (vec![local(1), remote(2), remote(3), FileInfo::default()], true),
+        ] {
+            let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+                .expect("the read-quorum side of a same-time transition split is served");
 
-        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
+            assert_eq!(selected.is_remote(), expect_remote);
+        }
+
+        for (tied, expect_remote) in [
+            (vec![local(1), local(2), remote(3), remote(4)], false),
+            (vec![remote(1), local(2), local(3), remote(4)], true),
+        ] {
+            let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; tied.len()], &tied, &[None, None, None, None], 2)
+                .expect("a same-time half split serves the side on the lowest-index disk");
+            assert_eq!(selected.is_remote(), expect_remote);
+        }
     }
 
     #[test]
-    fn test_latest_fileinfo_selection_rejects_same_modtime_transition_split_without_write_quorum() {
-        let mod_time = OffsetDateTime::now_utc();
-        let data_dir = Uuid::new_v4();
-        let old_meta_a = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 1);
-        let old_meta_b = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 2);
-        let mut partial_meta = quorum_test_fileinfo(mod_time, data_dir, "part-etag", 3);
-        partial_meta.transition_status = TRANSITION_COMPLETE.to_string();
-        partial_meta.transition_tier = "WARM".to_string();
-        partial_meta.transitioned_objname = "remote/object".to_string();
-        partial_meta.transition_version_id = Some(Uuid::new_v4());
-        let metas = vec![old_meta_a, old_meta_b, partial_meta, FileInfo::default()];
-        let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
-
-        let quorum = SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3);
-        let result = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, "", 2, 3);
-
-        assert_eq!(quorum, 2);
-        assert!(matches!(result, Err(DiskError::ErasureReadQuorum)));
-    }
-
-    #[test]
-    fn test_latest_fileinfo_selection_quorum_uses_write_quorum_for_degraded_committed_identity() {
+    fn test_latest_fileinfo_selection_serves_degraded_committed_identity() {
         let mod_time = OffsetDateTime::now_utc();
         let data_dir = Uuid::new_v4();
         let metas = vec![
@@ -12354,7 +12533,10 @@ mod tests {
         ];
         let errs = vec![None, None, None, Some(DiskError::DiskNotFound)];
 
-        assert_eq!(SetDisks::latest_fileinfo_selection_quorum("", &metas, &errs, 2, 3), 3);
+        let (_, selected) = SetDisks::select_valid_fileinfo(&vec![None; metas.len()], &metas, &errs, 2)
+            .expect("a committed identity stays readable with one disk missing");
+
+        assert_eq!(selected.parts[0].etag, "part-etag-a");
     }
 
     #[test]

crates/ecstore/src/set_disk/ops/object.rs

@@ -8839,7 +8839,7 @@ impl SetDisks {
         // quorum, failing write quorum on update_object_meta (backlog#872).
         let read_opts = opts.as_read_opts();
         let (mut fi, _, disks) = self
-            .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
+            .get_object_fileinfo_for_metadata_update(bucket, object, &read_opts)
             .await?
             .into_owned();
 
@@ -9294,36 +9294,56 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             }
         };
 
-        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) {
-            Ok((r, w)) => (
-                usize::try_from(r)
-                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![src_bucket, src_object]))?,
-                usize::try_from(w)
-                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![src_bucket, src_object]))?,
-            ),
+        let src_version_id = src_opts.version_id.as_deref().unwrap_or_default();
+        let quorum = Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count);
+        let selection = match &quorum {
+            Ok((read_quorum, write_quorum)) => {
+                let read_quorum = usize::try_from(*read_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![src_bucket, src_object]))?;
+                let write_quorum = usize::try_from(*write_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![src_bucket, src_object]))?;
+                Self::select_fileinfo_for_read(src_bucket, &disks, &metas, &errs, src_version_id, read_quorum, write_quorum)
+                    .map(|(online_disks, fi)| (online_disks, fi, write_quorum))
+            }
+            Err(err) => Err(err.clone()),
+        };
+        let (online_disks, mut fi, write_quorum) = match selection {
+            Ok(selection) => selection,
             Err(mut err) => {
-                if err == DiskError::ErasureReadQuorum
-                    && !src_bucket.starts_with(RUSTFS_META_BUCKET)
-                    && self
-                        .delete_if_dangling(src_bucket, src_object, &metas, &errs, &HashMap::new(), src_opts.clone())
-                        .await
-                        .is_ok()
+                // The merged latest version can be one a failed PUT left below
+                // quorum; resolve it like a latest GET before the dangling
+                // cleanup below can report a key with a committed version absent.
+                if let Some(snapshot) = self
+                    .select_latest_past_dangling_versions(src_bucket, src_object, src_opts, true, &disks, &err)
+                    .await
                 {
-                    if src_opts.version_id.is_some() {
-                        err = DiskError::FileVersionNotFound
-                    } else {
-                        err = DiskError::FileNotFound
+                    // A version-only copy writes the served version's own
+                    // copies, so they and its write quorum replace the latest
+                    // read's.
+                    let (fi, parts_metadata, online_disks) = snapshot.into_owned();
+                    metas = parts_metadata;
+                    let write_quorum = fi.write_quorum(self.default_write_quorum());
+                    (online_disks, fi, write_quorum)
+                } else {
+                    if quorum.is_err()
+                        && err == DiskError::ErasureReadQuorum
+                        && !src_bucket.starts_with(RUSTFS_META_BUCKET)
+                        && self
+                            .delete_if_dangling(src_bucket, src_object, &metas, &errs, &HashMap::new(), src_opts.clone())
+                            .await
+                            .is_ok()
+                    {
+                        if src_opts.version_id.is_some() {
+                            err = DiskError::FileVersionNotFound
+                        } else {
+                            err = DiskError::FileNotFound
+                        }
                     }
+                    return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
                 }
-                return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
             }
         };
 
-        let src_version_id = src_opts.version_id.as_deref().unwrap_or_default();
-        let (online_disks, mut fi, _) =
-            Self::select_valid_fileinfo(&disks, &metas, &errs, src_version_id, read_quorum, write_quorum)
-                .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
-
         if fi.deleted {
             if src_opts.version_id.is_none() {
                 return Err(to_object_err(Error::FileNotFound, vec![src_bucket, src_object]));
@@ -9461,6 +9481,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 }
             }
 
+            // A version-only copy writes and then reverts on a quorum miss;
+            // refuse it before writing when the served version's disks cannot
+            // reach its write quorum even if every write succeeds.
+            let reachable = online_disks
+                .iter()
+                .map(|disk| disk.is_none().then_some(DiskError::DiskNotFound))
+                .collect::<Vec<_>>();
+            if let Some(err) = reduce_write_quorum_errs(&reachable, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+                return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
+            }
             Self::write_unique_file_info(&online_disks, "", src_bucket, src_object, &metas, write_quorum)
                 .await
                 .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
@@ -11147,44 +11177,63 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             }
         };
 
-        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) {
-            Ok((read_quorum, write_quorum)) => (read_quorum, write_quorum),
+        let version_id = opts.version_id.as_deref().unwrap_or_default();
+        let quorum = Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count);
+        let selection = match &quorum {
+            Ok((read_quorum, write_quorum)) => {
+                let read_quorum = usize::try_from(*read_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
+                let write_quorum = usize::try_from(*write_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
+                Self::select_fileinfo_for_read(bucket, &disks, &metas, &errs, version_id, read_quorum, write_quorum)
+            }
+            Err(err) => Err(err.clone()),
+        };
+        let (online_disks, mut fi) = match selection {
+            Ok(selection) => selection,
             Err(mut err) => {
-                // A replication terminal CAS is a validation/write-back
-                // attempt, never a heal operation. In particular,
-                // ValidateOnly must remain zero-write, and a worker whose
-                // outer replication lease is already lost must not delete the
-                // surviving xl.meta copies before the namespace fence below
-                // can reject it. Leave dangling cleanup to heal/scanner.
-                if opts.replication_status_writeback.is_some() {
+                // The merged latest version can be one a failed PUT left below
+                // quorum; resolve it like a latest GET before the dangling
+                // cleanup below can report a key with a committed version absent.
+                // A replication terminal CAS stays side-effect free: the walk
+                // queues heal for the versions it skips, so it does not run.
+                let resolved = if opts.replication_status_writeback.is_none() {
+                    self.select_latest_past_dangling_versions(bucket, object, opts, false, &disks, &err)
+                        .await
+                } else {
+                    None
+                };
+                if let Some(snapshot) = resolved {
+                    let (fi, _, online_disks) = snapshot.into_owned();
+                    (online_disks, fi)
+                } else {
+                    // A replication terminal CAS is a validation/write-back
+                    // attempt, never a heal operation. In particular,
+                    // ValidateOnly must remain zero-write, and a worker whose
+                    // outer replication lease is already lost must not delete
+                    // the surviving xl.meta copies before the namespace fence
+                    // below can reject it. Leave dangling cleanup to
+                    // heal/scanner.
+                    if quorum.is_err()
+                        && opts.replication_status_writeback.is_none()
+                        && err == DiskError::ErasureReadQuorum
+                        && !bucket.starts_with(RUSTFS_META_BUCKET)
+                        && self
+                            .delete_if_dangling(bucket, object, &metas, &errs, &HashMap::new(), opts.clone())
+                            .await
+                            .is_ok()
+                    {
+                        if opts.version_id.is_some() {
+                            err = DiskError::FileVersionNotFound
+                        } else {
+                            err = DiskError::FileNotFound
+                        }
+                    }
                     return Err(to_object_err(err.into(), vec![bucket, object]));
                 }
-                if err == DiskError::ErasureReadQuorum
-                    && !bucket.starts_with(RUSTFS_META_BUCKET)
-                    && self
-                        .delete_if_dangling(bucket, object, &metas, &errs, &HashMap::new(), opts.clone())
-                        .await
-                        .is_ok()
-                {
-                    if opts.version_id.is_some() {
-                        err = DiskError::FileVersionNotFound
-                    } else {
-                        err = DiskError::FileNotFound
-                    }
-                }
-                return Err(to_object_err(err.into(), vec![bucket, object]));
             }
         };
 
-        let read_quorum =
-            usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
-        let write_quorum = usize::try_from(write_quorum)
-            .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
-
-        let version_id = opts.version_id.as_deref().unwrap_or_default();
-        let (online_disks, mut fi, _) = Self::select_valid_fileinfo(&disks, &metas, &errs, version_id, read_quorum, write_quorum)
-            .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
-
         if fi.deleted {
             return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object]));
         }

crates/ecstore/src/set_disk/read.rs

@@ -264,6 +264,11 @@ enum MetadataReadPurpose {
     GetObjectReader,
 }
 
+/// Upper bound on exact-version reads a latest read spends looking past
+/// dangling versions before it keeps the read-quorum error; each read is one
+/// metadata fan-out.
+const MAX_DANGLING_LATEST_VERSIONS: usize = 8;
+
 /// How a metadata read-quorum failure is reported when the replies prove the
 /// object or version dangling. Only the object read entry points
 /// (`get_object_info`, `get_object_reader` and prepared GET metadata) report
@@ -273,6 +278,10 @@ enum MetadataReadPurpose {
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum DanglingRead {
     KeepQuorumError,
+    /// Metadata updates of the latest version keep the quorum error but, like
+    /// object reads, look past newer versions proven dangling
+    /// (`read_latest_past_dangling_versions`).
+    SkipDanglingVersions,
     ReportNotFound,
 }
 
@@ -724,9 +733,10 @@ impl SetDisks {
     }
 
     /// Like `get_object_fileinfo`, but `allow_early_stop=false` forces the full
-    /// quorum fanout. Read-before-write callers (object tagging) must use this:
-    /// the returned online-disk set is the write target, and the early-stop
-    /// subset would fail write quorum (backlog#872 regression fix).
+    /// quorum fanout. Read-before-write callers must use this or
+    /// `get_object_fileinfo_for_metadata_update`: the returned online-disk set
+    /// is the write target, and the early-stop subset would fail write quorum
+    /// (backlog#872 regression fix).
     #[tracing::instrument(level = "debug", skip(self))]
     pub(in crate::set_disk) async fn get_object_fileinfo_gated(
         &self,
@@ -748,6 +758,64 @@ impl SetDisks {
         .await
     }
 
+    /// The full-fanout read behind a metadata update such as object tagging.
+    /// A latest read whose newest versions never committed resolves to the
+    /// version a latest GET serves, while a dangling key keeps the quorum
+    /// error. The returned online disks are the write target.
+    #[tracing::instrument(level = "debug", skip(self))]
+    pub(in crate::set_disk) async fn get_object_fileinfo_for_metadata_update(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+    ) -> Result<GetObjectFileInfo> {
+        self.get_object_fileinfo_gated_inner(
+            bucket,
+            object,
+            opts,
+            false,
+            false,
+            MetadataReadPurpose::FullLayout,
+            DanglingRead::SkipDanglingVersions,
+        )
+        .await
+    }
+
+    /// Recovers a failed latest selection in read-before-write callers that
+    /// read and select metadata themselves (`put_object_metadata` and the
+    /// CopyObject source) the way a latest GET does. `None` keeps the caller's
+    /// error: a failure other than read quorum, an explicit version, the
+    /// system bucket's commit-proof rule, or a walk that finds no readable
+    /// version or cannot read every drive. The selected metadata keeps its
+    /// part checksums because a version-only copy carries them into the new
+    /// version.
+    pub(in crate::set_disk) async fn select_latest_past_dangling_versions(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        read_data: bool,
+        disks: &[Option<DiskStore>],
+        err: &DiskError,
+    ) -> Option<GetObjectFileInfo> {
+        if *err != DiskError::ErasureReadQuorum || opts.version_id.is_some() || crate::bucket::utils::is_meta_bucketname(bucket) {
+            return None;
+        }
+        let opts = ObjectOptions {
+            include_part_checksums: true,
+            ..opts.clone()
+        };
+        Box::pin(self.read_latest_past_dangling_versions(
+            bucket,
+            object,
+            &opts,
+            read_data,
+            MetadataReadPurpose::FullLayout,
+            disks,
+        ))
+        .await
+    }
+
     #[allow(clippy::too_many_arguments)]
     async fn get_object_fileinfo_gated_inner(
         &self,
@@ -966,10 +1034,23 @@ impl SetDisks {
             return Err(to_object_err(err.into(), vec![bucket, object, &vid]));
         }
 
-        let (op_online_disks, mut fi, fileinfo_selection_quorum) =
-            match Self::select_valid_fileinfo(&disks, &parts_metadata, &errs, vid.as_str(), read_quorum, write_quorum) {
+        let (op_online_disks, mut fi) =
+            match Self::select_fileinfo_for_read(bucket, &disks, &parts_metadata, &errs, vid.as_str(), read_quorum, write_quorum)
+            {
                 Ok(selection) => selection,
                 Err(err) => {
+                    if err == DiskError::ErasureReadQuorum
+                        && vid.is_empty()
+                        && dangling_read != DanglingRead::KeepQuorumError
+                        && metadata_fanout_complete
+                        && !crate::bucket::utils::is_meta_bucketname(bucket)
+                        && let Some(snapshot) =
+                            Box::pin(self.read_latest_past_dangling_versions(bucket, object, opts, read_data, purpose, &disks))
+                                .await
+                    {
+                        rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, "reached");
+                        return Ok(snapshot);
+                    }
                     let err = StorageError::from(err);
                     rustfs_io_metrics::record_get_object_metadata_quorum_result(
                         metadata_metrics_path,
@@ -987,7 +1068,7 @@ impl SetDisks {
                 rustfs_utils::http::remove_str(&mut metadata.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
             }
         }
-        metadata_fanout_diagnostics.record_quorum_candidate_latency(metadata_metrics_path, fileinfo_selection_quorum);
+        metadata_fanout_diagnostics.record_quorum_candidate_latency(metadata_metrics_path, read_quorum);
         let needs_late_metadata_fanout = !metadata_fanout_complete
             && allow_early_stop
             && non_inline_data_read_early_stop_allowed(read_data, bucket, object)
@@ -1014,7 +1095,6 @@ impl SetDisks {
                     && allow_early_stop
                     && metadata_fanout_diagnostics.valid_responses() >= read_quorum
                     && errs.iter().all(Option::is_none)
-                    && fileinfo_selection_quorum >= read_quorum
                     && op_online_disks.iter().filter(|disk| disk.is_some()).count() >= read_quorum))
         {
             #[cfg(test)]
@@ -1051,6 +1131,110 @@ impl SetDisks {
         Ok(snapshot)
     }
 
+    /// A latest read whose split cannot reach read quorum may still have a
+    /// readable older version under newer versions that never committed;
+    /// MinIO resolves this in `pickLatestQuorumFilesInfo`. With every drive
+    /// answering, try versions newest first through exact-version reads and
+    /// serve the readable one with the newest selected modification time. A
+    /// version whose exact read proves it dangling (the NoSuchVersion an
+    /// explicit GET reports) is skipped and queued for heal, which owns its
+    /// cleanup; any other outcome keeps the read-quorum error.
+    async fn read_latest_past_dangling_versions(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        read_data: bool,
+        purpose: MetadataReadPurpose,
+        disks: &[Option<DiskStore>],
+    ) -> Option<GetObjectFileInfo> {
+        let disk_object = rustfs_utils::path::encode_dir_object(object);
+        let (raw, errs) = Self::read_all_raw_file_info(disks, bucket, &disk_object, false).await;
+        if errs
+            .iter()
+            .any(|err| err.as_ref().is_some_and(|err| !matches!(err, DiskError::FileNotFound)))
+        {
+            return None;
+        }
+        // Stacks are newest first, so only each drive's first few headers can
+        // matter. Keep a version's newest modification time across drives: a
+        // rewritten null version can carry different times on different drives.
+        let mut newest = std::collections::HashMap::<Option<uuid::Uuid>, Option<time::OffsetDateTime>>::new();
+        let mut truncated = false;
+        for raw in raw.into_iter().flatten() {
+            let meta = rustfs_filemeta::FileMeta::load(&raw.buf).ok()?;
+            let mut versions = meta.versions.iter().filter(|version| !version.header.free_version());
+            for version in versions.by_ref().take(MAX_DANGLING_LATEST_VERSIONS) {
+                let mod_time = newest.entry(version.header.version_id).or_insert(version.header.mod_time);
+                *mod_time = (*mod_time).max(version.header.mod_time);
+            }
+            truncated |= versions.next().is_some();
+        }
+        let mut candidates = newest.into_iter().collect::<Vec<_>>();
+        candidates.sort_by(|left, right| right.1.cmp(&left.1).then_with(|| left.0.cmp(&right.0)));
+
+        let mut best: Option<(Option<time::OffsetDateTime>, GetObjectFileInfo)> = None;
+        let mut skipped = Vec::new();
+        let mut settled = false;
+        for (reads, (version_id, newest_mod_time)) in candidates.into_iter().enumerate() {
+            if best.as_ref().is_some_and(|(selected, _)| newest_mod_time <= *selected) {
+                settled = true;
+                break;
+            }
+            if reads == MAX_DANGLING_LATEST_VERSIONS {
+                return None;
+            }
+            // The nil UUID addresses the null version slot at this layer.
+            let version_id = version_id.unwrap_or_else(uuid::Uuid::nil).to_string();
+            let exact = ObjectOptions {
+                version_id: Some(version_id.clone()),
+                ..opts.clone()
+            };
+            match Box::pin(self.get_object_fileinfo_gated_inner(
+                bucket,
+                object,
+                &exact,
+                read_data,
+                false,
+                purpose,
+                DanglingRead::ReportNotFound,
+            ))
+            .await
+            {
+                Ok(snapshot) => {
+                    let selected = snapshot.fi().mod_time;
+                    if best.as_ref().is_none_or(|(current, _)| selected > *current) {
+                        best = Some((selected, snapshot));
+                    }
+                }
+                Err(StorageError::VersionNotFound(..)) => skipped.push(version_id),
+                Err(_) => return None,
+            }
+        }
+        // Headers past each drive's first few were not compared; only a walk
+        // that proved no remaining candidate can be newer may stop early.
+        if !settled && truncated {
+            return None;
+        }
+        let (_, mut snapshot) = best?;
+        snapshot.mark_selected_as_latest();
+        if !opts.suppress_read_repair {
+            for version_id in &skipped {
+                submit_read_repair_heal(
+                    bucket,
+                    object,
+                    Some(version_id),
+                    self.pool_index,
+                    self.set_index,
+                    None,
+                    "dangling_latest_version",
+                )
+                .await;
+            }
+        }
+        Some(snapshot)
+    }
+
     #[hotpath::measure(impl_type = "SetDisks")]
     pub(super) async fn get_object_info_fileinfo_and_quorum(
         &self,
@@ -2465,8 +2649,8 @@ impl SetDisks {
             return Err(to_object_err(err.into(), vec![bucket, object]));
         }
 
-        let (mut online_disks, full_fi, _) =
-            SetDisks::select_valid_fileinfo(fallback_disks, &parts_metadata, &errs, "", read_quorum, write_quorum)?;
+        let (mut online_disks, full_fi) =
+            SetDisks::select_fileinfo_for_read(bucket, fallback_disks, &parts_metadata, &errs, "", read_quorum, write_quorum)?;
         if !late_metadata_read_identity_matches(expected, &full_fi) {
             return Err(to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]));
         }
@@ -4052,8 +4236,7 @@ mod metadata_cache_tests {
         assert!(original.is_complete());
         let (baseline, errors, _) = original.into_legacy();
         assert!(errors.iter().all(Option::is_none), "original real shards must all be readable");
-        let (_, selected, _) =
-            SetDisks::select_valid_fileinfo(&disks, &baseline, &errors, "", 2, 3).expect("original metadata quorum");
+        let (_, selected) = SetDisks::select_valid_fileinfo(&disks, &baseline, &errors, 2).expect("original metadata quorum");
         assert!(!selected.inline_data(), "late fixture must retain real non-inline shard files");
         let expected = LateMetadataIdentity::from_file_info(&selected);
         let (_, _, original_online) =
@@ -4086,7 +4269,7 @@ mod metadata_cache_tests {
                 .expect("read changed real disk metadata");
             assert!(changed.is_complete(), "late fixture must reach complete metadata observation");
             let (metadata, errors, _) = changed.into_legacy();
-            let (online, selected, _) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errors, "", 2, 3)
+            let (online, selected) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errors, 2)
                 .expect("changed metadata still reaches selection quorum");
             assert_eq!(online.iter().flatten().count(), 4, "the late identity guard must decide this failure");
             if changed_generation {
@@ -5135,7 +5318,7 @@ mod metadata_cache_tests {
         // This seam prepares a real, independently validated D1/P1/P2 quorum.
         // The hedge test separately exercises its production publication timing.
         let errs: Vec<Option<DiskError>> = (0..4).map(|_| None).collect();
-        let (online_disks, fi, _) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errs, "", 2, 3)
+        let (online_disks, fi) = SetDisks::select_valid_fileinfo(&disks, &metadata, &errs, 2)
             .expect("three real members satisfy latest-generation selection");
         assert_eq!(online_disks.iter().flatten().count(), 3);
         assert_eq!(fi.metadata.get("etag"), committed.etag.as_ref());
@@ -6240,16 +6423,17 @@ mod tests {
         let parts_metadata = vec![first.clone(), second, third, FileInfo::default()];
         let errs = vec![None, None, None, None];
 
-        let (read_quorum, write_quorum) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 2)
+        let (read_quorum, _) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 2)
             .expect("partial early-stop metadata should preserve read quorum");
         let read_quorum = usize::try_from(read_quorum).expect("read quorum should be non-negative");
-        let write_quorum = usize::try_from(write_quorum).expect("write quorum should be non-negative");
-        let selection_quorum = SetDisks::latest_fileinfo_selection_quorum("", &parts_metadata, &errs, read_quorum, write_quorum);
+        let (_, latest) = SetDisks::select_valid_fileinfo(&vec![None; parts_metadata.len()], &parts_metadata, &errs, read_quorum)
+            .expect("partial early-stop metadata should select the candidate");
         let selected = SetDisks::pick_valid_fileinfo(&parts_metadata, None, Some("etag-1".to_string()), read_quorum)
             .expect("partial early-stop metadata should preserve selected FileInfo");
 
         assert_eq!(read_quorum, 2);
-        assert_eq!(selection_quorum, 3);
+        assert_eq!(latest.name, first.name);
+        assert_eq!(latest.get_etag(), first.get_etag());
         assert_eq!(selected.name, first.name);
         assert_eq!(selected.get_etag(), first.get_etag());
     }
@@ -9751,6 +9935,7 @@ mod dangling_read_tests {
     use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
     use crate::storage_api_contracts::object::ObjectOperations;
     use http::HeaderMap;
+    use std::collections::HashMap;
     use std::path::PathBuf;
     use tempfile::TempDir;
     use tokio::io::AsyncReadExt;
@@ -9761,9 +9946,13 @@ mod dangling_read_tests {
     const PARITY: usize = 2;
 
     async fn dangling_test_set(bucket: &str) -> (Vec<TempDir>, Arc<SetDisks>) {
+        dangling_test_set_with(bucket, DRIVES, PARITY).await
+    }
+
+    async fn dangling_test_set_with(bucket: &str, drives: usize, parity: usize) -> (Vec<TempDir>, Arc<SetDisks>) {
         let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
         ctx.update_erasure_type(crate::layout::endpoints::SetupType::Erasure).await;
-        let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(DRIVES, PARITY, ctx).await;
+        let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(drives, parity, ctx).await;
         set.make_bucket(bucket, &MakeBucketOptions::default())
             .await
             .expect("test bucket should be created on every drive");
@@ -9777,6 +9966,9 @@ mod dangling_read_tests {
             &mut PutObjReader::from_vec(body.to_vec()),
             &ObjectOptions {
                 versioned,
+                // Wait for every drive so tests that edit one drive's metadata
+                // do not race the PUT's background tail writes.
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )
@@ -9796,6 +9988,25 @@ mod dangling_read_tests {
         }
     }
 
+    /// Drops `version_id` from `disks`, modelling drives a failed PUT never
+    /// reached while the older versions below it stay intact.
+    async fn drop_version_from(set: &SetDisks, bucket: &str, object: &str, version_id: Uuid, disks: std::ops::Range<usize>) {
+        let all = set.disks.read().await.clone();
+        for disk in &all[disks] {
+            let disk = disk.as_ref().expect("every drive should be online");
+            let mut remove = FileInfo {
+                volume: bucket.to_string(),
+                name: object.to_string(),
+                version_id: Some(version_id),
+                ..Default::default()
+            };
+            remove.set_skip_tier_free_version();
+            disk.delete_version(bucket, object, remove, false, DeleteOptions::default())
+                .await
+                .expect("dropping the version should model a drive that never received it");
+        }
+    }
+
     fn read_opts(version_id: Option<&str>) -> ObjectOptions {
         ObjectOptions {
             version_id: version_id.map(str::to_owned),
@@ -9955,12 +10166,539 @@ mod dangling_read_tests {
         head(&set, bucket, object, Some(&v1))
             .await
             .expect("HEAD of the committed version must succeed");
-        // The latest read sees two generations (version two on three drives,
-        // version one on the rest), which is not proof of absence.
-        let Err(err) = head(&set, bucket, object, None).await else {
-            panic!("a split latest version must not resolve without quorum");
+        // The latest read sees version two on three drives and version one on
+        // the rest. Version two's exact read proves it dangling, so the latest
+        // read serves version one, the newest version an exact read can serve.
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("the latest read must serve the newest readable version");
+        assert_eq!(latest.version_id.map(|id| id.to_string()).as_deref(), Some(v1.as_str()));
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"version one");
+    }
+
+    #[tokio::test]
+    async fn latest_read_skips_dangling_versions_above_a_committed_version() {
+        // EC 12+4: a failed PUT that left version two on 5..=11 drives keeps
+        // version one, present in every stack, below its read quorum of twelve
+        // when only the top of each stack is read. Up to seven copies the
+        // listing quorum already shows version one; from eight it lists the
+        // dangling version two.
+        let bucket = "dangling-latest-12-4";
+        let (_dirs, set) = dangling_test_set_with(bucket, 16, 4).await;
+        for copies in 5..=11 {
+            let object = format!("torn/latest-{copies:02}");
+            let v1 = put(&set, bucket, &object, b"committed version", true)
+                .await
+                .version_id
+                .expect("versioned PUT should assign a version id");
+            let v2 = put(&set, bucket, &object, b"failed overwrite", true)
+                .await
+                .version_id
+                .expect("versioned PUT should assign a version id");
+            drop_version_from(&set, bucket, &object, v2, copies..16).await;
+
+            let context = format!("version two on {copies} of 16 drives");
+            let latest = head(&set, bucket, &object, None)
+                .await
+                .unwrap_or_else(|err| panic!("{context}: HEAD must serve version one: {err:?}"));
+            assert_eq!(latest.version_id, Some(v1), "{context}");
+            assert!(latest.is_latest, "{context}: the served version is the latest one");
+            assert_eq!(
+                get(&set, bucket, &object, None)
+                    .await
+                    .unwrap_or_else(|err| panic!("{context}: GET must serve version one: {err:?}")),
+                b"committed version",
+                "{context}"
+            );
+            let Err(err) = head(&set, bucket, &object, Some(&v2.to_string())).await else {
+                panic!("{context}: the dangling version must not be served");
+            };
+            assert!(matches!(err, StorageError::VersionNotFound(..)), "{context}: {err:?}");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_read_keeps_quorum_error_when_the_newest_version_is_uncertain() {
+        // Version two is on every drive, but a torn metadata update left two
+        // identities at one modification time, three copies each. Its exact
+        // read fails with a quorum error rather than NoSuchVersion, so the
+        // walk must not fall through to version one.
+        let bucket = "dangling-latest-uncertain";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        put(&set, bucket, object, b"version one", true).await;
+        let v2 = put(&set, bucket, object, b"version two", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id")
+            .to_string();
+        let disks = set.disks.read().await.clone();
+        for disk in disks.iter().take(3) {
+            let disk = disk.as_ref().expect("every drive should be online");
+            let mut fi = disk
+                .read_version("", bucket, object, &v2, &ReadOptions::default())
+                .await
+                .expect("version two should be on every drive");
+            fi.metadata.insert("x-amz-meta-torn".to_string(), "half-applied".to_string());
+            disk.update_metadata(bucket, object, fi, &crate::disk::UpdateMetadataOpts::default())
+                .await
+                .expect("a half-applied metadata update should be stored");
+        }
+
+        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "torn metadata update on version two").await;
+    }
+
+    #[tokio::test]
+    async fn latest_read_serves_a_committed_null_version_below_a_dangling_version() {
+        // A null version written while versioning was suspended sits between
+        // version one and a dangling version three. The walk reaches the null
+        // slot through the nil UUID and serves it, not version one.
+        let bucket = "dangling-latest-null";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let v1 = put(&set, bucket, object, b"version one", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        set.put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(b"null version".to_vec()),
+            &ObjectOptions {
+                version_suspended: true,
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("a suspended-versioning PUT should commit the null version");
+        let v3 = put(&set, bucket, object, b"version three", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(&set, bucket, object, v3, 3..DRIVES).await;
+
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("the committed null version must be served");
+        assert_ne!(latest.version_id, Some(v1));
+        assert_ne!(latest.version_id, Some(v3));
+        assert!(latest.is_latest);
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"null version");
+    }
+
+    #[tokio::test]
+    async fn latest_read_orders_versions_by_their_newest_modification_time() {
+        // Drive 0 missed the suspended-versioning rewrite of the null version,
+        // so its null copy predates version one while every other drive holds
+        // the committed rewrite. Ordering by the first drive's time would try
+        // version one before the null version and serve it.
+        let bucket = "dangling-latest-order";
+        let object = "torn/latest";
+        let (dirs, set) = dangling_test_set(bucket).await;
+        let suspended = ObjectOptions {
+            version_suspended: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
+            ..Default::default()
         };
-        assert!(is_read_quorum_error(&err), "a split latest version keeps the quorum error: {err:?}");
+        set.put_object(bucket, object, &mut PutObjReader::from_vec(b"stale null".to_vec()), &suspended)
+            .await
+            .expect("the first null version should commit");
+        let v1 = put(&set, bucket, object, b"version one", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        let drive_zero = object_dir(&dirs, 0, bucket, object).join(STORAGE_FORMAT_FILE);
+        let stale = tokio::fs::read(&drive_zero)
+            .await
+            .expect("drive zero metadata should be readable");
+        set.put_object(bucket, object, &mut PutObjReader::from_vec(b"committed null".to_vec()), &suspended)
+            .await
+            .expect("the null rewrite should commit");
+        tokio::fs::write(&drive_zero, stale)
+            .await
+            .expect("drive zero should keep the pre-rewrite stack");
+        let v5 = put(&set, bucket, object, b"version five", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(&set, bucket, object, v5, 3..DRIVES).await;
+
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("the committed null rewrite must be served");
+        assert_ne!(latest.version_id, Some(v1));
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"committed null");
+    }
+
+    #[tokio::test]
+    async fn latest_read_past_dangling_versions_requires_every_drive_to_answer() {
+        let bucket = "dangling-latest-offline";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        put(&set, bucket, object, b"version one", true).await;
+        let v2 = put(&set, bucket, object, b"version two", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(&set, bucket, object, v2, 3..DRIVES).await;
+        // The offline drive may hold the copy that makes version two readable.
+        let offline = set.disks.write().await[5].take().expect("drive five should start online");
+
+        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "one drive offline").await;
+
+        set.disks.write().await[5] = Some(offline);
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("with every drive answering the dangling version is skipped");
+        assert_ne!(latest.version_id, Some(v2));
+    }
+
+    /// Version one on every drive under a version two that a failed PUT left
+    /// on drives `0..3`, below the read quorum of four.
+    async fn put_committed_below_dangling(set: &SetDisks, bucket: &str, object: &str) -> (Uuid, Uuid) {
+        let v1 = put(set, bucket, object, b"version one", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        let v2 = put(set, bucket, object, b"version two", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(set, bucket, object, v2, 3..DRIVES).await;
+        (v1, v2)
+    }
+
+    /// Each drive's copy of `version_id`, or `None` where the drive lacks it.
+    async fn version_metadata(
+        set: &SetDisks,
+        bucket: &str,
+        object: &str,
+        version_id: Uuid,
+    ) -> Vec<Option<HashMap<String, String>>> {
+        let disks = set.disks.read().await.clone();
+        let mut copies = Vec::with_capacity(disks.len());
+        for disk in &disks {
+            let disk = disk.as_ref().expect("every drive should be online");
+            copies.push(
+                disk.read_version("", bucket, object, &version_id.to_string(), &ReadOptions::default())
+                    .await
+                    .ok()
+                    .map(|fi| fi.metadata),
+            );
+        }
+        copies
+    }
+
+    #[tokio::test]
+    async fn latest_tagging_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-tagging";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, v2) = put_committed_below_dangling(&set, bucket, object).await;
+        let tag_of = |metadata: &Option<HashMap<String, String>>| {
+            metadata
+                .as_ref()
+                .and_then(|metadata| metadata.get(rustfs_utils::http::headers::AMZ_OBJECT_TAGGING).cloned())
+        };
+
+        let tagged = set
+            .put_object_tags(bucket, object, "stage=committed", &ObjectOptions::default())
+            .await
+            .expect("PutObjectTagging must update the version a latest GET serves");
+        assert_eq!(tagged.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert_eq!(
+                tag_of(metadata).as_deref(),
+                Some("stage=committed"),
+                "drive {disk} must carry the tag on version one"
+            );
+        }
+        for (disk, metadata) in version_metadata(&set, bucket, object, v2).await.iter().enumerate() {
+            assert_eq!(tag_of(metadata), None, "drive {disk} must leave the dangling version untouched");
+        }
+        assert_eq!(*head(&set, bucket, object, None).await.expect("latest HEAD").user_tags, "stage=committed");
+
+        let untagged = set
+            .delete_object_tags(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("DeleteObjectTagging must update the version a latest GET serves");
+        assert_eq!(untagged.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert_eq!(tag_of(metadata).as_deref(), Some(""), "drive {disk} must clear the tag on version one");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_update_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-metadata";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, v2) = put_committed_below_dangling(&set, bucket, object).await;
+
+        let updated = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "updated".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a latest metadata update must target the version a latest GET serves");
+        assert_eq!(updated.version_id, Some(v1));
+        assert!(updated.is_latest, "the updated version is the latest one");
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert_eq!(metadata.get("x-amz-meta-probe").map(String::as_str), Some("updated"), "drive {disk}");
+        }
+        for (disk, metadata) in version_metadata(&set, bucket, object, v2).await.iter().enumerate() {
+            assert!(
+                metadata
+                    .as_ref()
+                    .is_none_or(|metadata| !metadata.contains_key("x-amz-meta-probe")),
+                "drive {disk} must leave the dangling version untouched"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_copy_source_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-copy";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+
+        // A metadata-only self-copy rewrites the latest version in place.
+        let object = "torn/in-place";
+        let (v1, _) = put_committed_below_dangling(&set, bucket, object).await;
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        assert_eq!(source.version_id, Some(v1));
+        source.metadata_only = true;
+        source.user_defined = Arc::new(HashMap::from([("x-amz-meta-copied".to_string(), "in-place".to_string())]));
+        let copied = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions::default(),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("a CopyObject source without a version id must resolve like a latest GET");
+        assert_eq!(copied.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert_eq!(metadata.get("x-amz-meta-copied").map(String::as_str), Some("in-place"), "drive {disk}");
+        }
+
+        // A version-only self-copy commits a new version that shares the
+        // served version's data.
+        let object = "torn/new-version";
+        let (_, v2) = put_committed_below_dangling(&set, bucket, object).await;
+        let v3 = Uuid::new_v4();
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        source.version_only = true;
+        let copied = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+                &ObjectOptions {
+                    version_id: Some(v3.to_string()),
+                    versioned: true,
+                    object_lock_config_snapshot: Some(Arc::new(crate::object_api::ObjectLockConfigSnapshot::new(
+                        crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+                    ))),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a version-only CopyObject source without a version id must resolve like a latest GET");
+        assert_eq!(copied.version_id, Some(v3));
+        let latest = head(&set, bucket, object, None).await.expect("latest HEAD");
+        assert_eq!(latest.version_id, Some(v3));
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"version one");
+        assert!(
+            version_metadata(&set, bucket, object, v2).await[..3]
+                .iter()
+                .all(Option::is_some),
+            "the copy must not remove the dangling version; heal owns its cleanup"
+        );
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_update_past_dangling_versions_keeps_write_quorum() {
+        // EC 3+3: read quorum three, write quorum four. Version one reached
+        // three drives, so it is readable but below write quorum; versions two
+        // and three are dangling on two drives each. A latest read serves
+        // version one, and a metadata update of it is refused before writing.
+        let bucket = "dangling-latest-write-quorum";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set_with(bucket, 6, 3).await;
+        let mut versions = Vec::new();
+        for body in [b"version zero".as_slice(), b"version one", b"version two", b"version three"] {
+            versions.push(
+                put(&set, bucket, object, body, true)
+                    .await
+                    .version_id
+                    .expect("versioned PUT should assign a version id"),
+            );
+        }
+        let [_, v1, v2, v3] = versions[..] else {
+            unreachable!("four versions were written");
+        };
+        drop_version_from(&set, bucket, object, v1, 3..6).await;
+        drop_version_from(&set, bucket, object, v2, 2..6).await;
+        drop_version_from(&set, bucket, object, v3, 0..3).await;
+        drop_version_from(&set, bucket, object, v3, 5..6).await;
+
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("version one is readable at read quorum");
+        assert_eq!(latest.version_id, Some(v1));
+
+        let is_write_quorum_error =
+            |err: &StorageError| matches!(err, StorageError::ErasureWriteQuorum | StorageError::InsufficientWriteQuorum(_, _));
+        let Err(err) = set
+            .put_object_tags(bucket, object, "stage=refused", &ObjectOptions::default())
+            .await
+        else {
+            panic!("tagging a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "tagging keeps the write-quorum error: {err:?}");
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "refused".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a metadata update of a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "the metadata update keeps the write-quorum error: {err:?}");
+        let v4 = Uuid::new_v4();
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        source.version_only = true;
+        let Err(err) = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+                &ObjectOptions {
+                    version_id: Some(v4.to_string()),
+                    versioned: true,
+                    object_lock_config_snapshot: Some(Arc::new(crate::object_api::ObjectLockConfigSnapshot::new(
+                        crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+                    ))),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a version-only copy of a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "the version-only copy keeps the write-quorum error: {err:?}");
+        assert!(
+            version_metadata(&set, bucket, object, v4).await.iter().all(Option::is_none),
+            "a refused version-only copy must not write the new version to any drive"
+        );
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert!(
+                metadata.as_ref().is_none_or(|metadata| {
+                    !metadata.contains_key(rustfs_utils::http::headers::AMZ_OBJECT_TAGGING)
+                        && !metadata.contains_key("x-amz-meta-probe")
+                }),
+                "drive {disk} must not be written by a refused update"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn replication_status_writeback_never_walks_past_dangling_versions() {
+        // A replication terminal CAS stays side-effect free: it keeps the
+        // read-quorum error instead of resolving the latest version, so it
+        // neither queues heal for the dangling version nor writes version one.
+        let bucket = "dangling-latest-replication-cas";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, _) = put_committed_below_dangling(&set, bucket, object).await;
+
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "written".to_string())])),
+                    replication_status_writeback: Some(Box::new(crate::object_api::ReplicationStatusWritebackCondition {
+                        mode: crate::object_api::ReplicationStatusWritebackMode::ValidateOnly,
+                        ..Default::default()
+                    })),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a replication terminal CAS must not resolve past a dangling version");
+        };
+        assert!(is_read_quorum_error(&err), "the CAS keeps the read-quorum error: {err:?}");
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert!(!metadata.contains_key("x-amz-meta-probe"), "drive {disk} must not be written");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_updates_past_dangling_versions_require_every_drive_to_answer() {
+        let bucket = "dangling-latest-update-offline";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        put_committed_below_dangling(&set, bucket, object).await;
+        let offline = set.disks.write().await[5].take().expect("drive five should start online");
+
+        let Err(err) = set
+            .put_object_tags(bucket, object, "stage=uncertain", &ObjectOptions::default())
+            .await
+        else {
+            panic!("tagging must not resolve past a version an offline drive may commit");
+        };
+        assert!(is_read_quorum_error(&err), "tagging keeps the read-quorum error: {err:?}");
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "uncertain".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a metadata update must not resolve past a version an offline drive may commit");
+        };
+        assert!(is_read_quorum_error(&err), "the metadata update keeps the read-quorum error: {err:?}");
+        set.disks.write().await[5] = Some(offline);
     }
 
     #[tokio::test]

crates/ecstore/src/store/list_objects.rs

@@ -22,9 +22,7 @@ fn to_filemeta_err(err: Error) -> rustfs_filemeta::Error {
 }
 
 use crate::bucket::lifecycle::bucket_lifecycle_ops::free_version_remote_tuple_matches;
-use crate::bucket::metadata_sys::{
-    get_versioning_config, has_authoritative_never_versioned_state, has_authoritative_never_versioned_state_in,
-};
+use crate::bucket::metadata_sys::{get_versioning_config, has_authoritative_never_versioned_state};
 use crate::bucket::utils::{check_list_objs_args, has_bad_path_component};
 use crate::bucket::versioning::VersioningApi;
 use crate::cache_value::metacache_set::{
@@ -373,49 +371,6 @@ async fn can_skip_hidden_prefix_check(options: &ListPathOptions) -> bool {
         .unwrap_or(false)
 }
 
-/// Whether an empty listing of `prefix` is the proof that lets the caller
-/// reclaim delete residue under it. Any first page that scanned the whole
-/// prefix without finding an object or a sub-prefix qualifies, with or without
-/// a delimiter: that is the shape clients issue when they stat, browse, or
-/// recursively remove a phantom folder. The purge itself re-verifies every
-/// directory on every disk before deleting anything.
-fn should_purge_empty_directory_listing(
-    prefix: &str,
-    marker: Option<&str>,
-    max_keys: i32,
-    incl_deleted: bool,
-    result: &ListObjectsInfo,
-) -> bool {
-    // list_path rejects leading slashes, so an empty result cannot justify cleanup.
-    !prefix.is_empty()
-        && !prefix.starts_with(SLASH_SEPARATOR)
-        && prefix.ends_with(SLASH_SEPARATOR)
-        && marker.is_none()
-        && max_keys > 0
-        && !incl_deleted
-        && !result.is_truncated
-        && result.objects.is_empty()
-        && result.prefixes.is_empty()
-}
-
-fn should_purge_empty_recursive_bucket_listing(
-    prefix: &str,
-    delimiter: Option<&str>,
-    marker: Option<&str>,
-    max_keys: i32,
-    incl_deleted: bool,
-    result: &ListObjectsInfo,
-) -> bool {
-    prefix.is_empty()
-        && delimiter.is_none()
-        && marker.is_none()
-        && max_keys > 0
-        && !incl_deleted
-        && !result.is_truncated
-        && result.objects.is_empty()
-        && result.prefixes.is_empty()
-}
-
 const MARKER_TAG_VERSION: &str = "v2";
 const LEGACY_MARKER_TAG_VERSIONS: &[&str] = &["v1", MARKER_TAG_VERSION];
 const LIST_CACHE_MARKER_PREFIX: &str = "[rustfs_cache:";
@@ -1022,39 +977,15 @@ impl ECStore {
         let get_objects =
             ObjectInfo::from_meta_cache_entries_sorted_infos(&page.entries, bucket, prefix, delimiter.clone()).await;
 
-        let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) =
+        let (objects, prefixes, is_truncated, next_marker, _) =
             list_objects_paginate(get_objects, &delimiter, max_keys, disk_has_more, false, last_scanned_key.as_deref());
-        let _ = next_version_idmarker;
 
-        let result = ListObjectsInfo {
+        Ok(ListObjectsInfo {
             is_truncated,
             next_marker,
             objects,
             prefixes,
-        };
-
-        let purge_exact_prefix =
-            should_purge_empty_directory_listing(prefix, opts.marker.as_deref(), max_keys, incl_deleted, &result);
-        let purge_empty_bucket = should_purge_empty_recursive_bucket_listing(
-            prefix,
-            delimiter.as_deref(),
-            opts.marker.as_deref(),
-            max_keys,
-            incl_deleted,
-            &result,
-        );
-        if (purge_exact_prefix || purge_empty_bucket)
-            && has_authoritative_never_versioned_state_in(&self.ctx, bucket)
-                .await
-                .unwrap_or(false)
-        {
-            if purge_exact_prefix {
-                self.purge_orphan_dir_object(bucket, prefix).await;
-            } else {
-                self.purge_orphan_dir_objects_in_bucket(bucket).await;
-            }
-        }
-        Ok(result)
+        })
     }
 
     pub async fn inner_list_object_versions(
@@ -2900,13 +2831,12 @@ mod lifecycle_exact_tests;
 #[cfg(test)]
 mod test {
     use super::{
-        ENV_API_LIST_OBJECTS_QUORUM, GatherResultsState, ListObjectsInfo, ListPathOptions, ListSourceMode, MAX_OBJECT_LIST,
-        RUSTFS_META_BUCKET, VersionMarker, gather_results, list_marker_key, list_merged_entry_channel,
-        list_metadata_resolution_params, list_objects_paginate, list_objects_quorum_from_env, max_keys_plus_one,
-        merge_entry_channels, normalize_list_quorum, observe_scanner_namespace_mutations, parse_version_marker,
-        recursive_walk_list_path_options, reset_scanner_namespace_mutation_generation_for_test,
-        scanner_namespace_mutation_generation, send_or_cancel, should_purge_empty_directory_listing, version_marker_for_entries,
-        walk_result_from_set_errors,
+        ENV_API_LIST_OBJECTS_QUORUM, GatherResultsState, ListPathOptions, ListSourceMode, MAX_OBJECT_LIST, RUSTFS_META_BUCKET,
+        VersionMarker, gather_results, list_marker_key, list_merged_entry_channel, list_metadata_resolution_params,
+        list_objects_paginate, list_objects_quorum_from_env, max_keys_plus_one, merge_entry_channels, normalize_list_quorum,
+        observe_scanner_namespace_mutations, parse_version_marker, recursive_walk_list_path_options,
+        reset_scanner_namespace_mutation_generation_for_test, scanner_namespace_mutation_generation, send_or_cancel,
+        version_marker_for_entries, walk_result_from_set_errors,
     };
     use crate::bucket::replication::ReplicationState;
     use crate::cache_value::metacache_set::ListPathRawOutcome;
@@ -4949,77 +4879,114 @@ mod test {
         assert_eq!(scanner_namespace_mutation_generation(), 2);
     }
 
-    #[test]
-    fn empty_directory_listing_purge_requires_complete_first_page_of_prefix() {
-        let empty = ListObjectsInfo::default();
-        assert!(should_purge_empty_directory_listing("ghost/", None, 1, false, &empty));
-        assert!(should_purge_empty_directory_listing("ghost/", None, 1000, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost", None, 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", Some("marker"), 1, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 0, false, &empty));
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, true, &empty));
+    #[tokio::test]
+    async fn listing_preserves_valid_legacy_residue_until_explicit_delete() {
+        use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
 
-        let mut live = ListObjectsInfo::default();
-        live.objects.push(ObjectInfo::default());
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &live));
+        use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
 
-        let mut prefixed = ListObjectsInfo::default();
-        prefixed.prefixes.push("ghost/child/".to_owned());
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &prefixed));
+        let (dirs, store) = isolated_store_over_temp_disks().await;
+        let bucket = "stage-e-exact-delete-legacy";
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created with authoritative metadata");
+        let transaction = uuid::Uuid::new_v4();
+        let mut backups = Vec::new();
+        for object in ["repro/obj-00001.bin", "repro/obj-00002.bin"] {
+            let mut reader = crate::object_api::PutObjReader::from_vec(vec![0x62; 1024 * 1024]);
+            store
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &crate::object_api::ObjectOptions {
+                        write_completion: crate::object_api::WriteCompletion::TailDrained,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("create genuine per-drive metadata and shards");
+            for dir in &dirs {
+                let object_dir = dir.path().join(bucket).join(object);
+                let mut entries = tokio::fs::read_dir(&object_dir).await.expect("read actual object root");
+                let mut marked = 0;
+                while let Some(entry) = entries.next_entry().await.expect("read object entry") {
+                    if uuid::Uuid::parse_str(&entry.file_name().to_string_lossy()).is_ok() {
+                        tokio::fs::write(
+                            entry
+                                .path()
+                                .join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction)),
+                            [],
+                        )
+                        .await
+                        .expect("mark actual data dir as historical delete residue");
+                        marked += 1;
+                    }
+                }
+                assert_eq!(marked, 1, "the real shard must have exactly one data directory");
+                let backup = object_dir.join(transaction.to_string());
+                tokio::fs::create_dir_all(&backup).await.expect("create legacy rollback root");
+                tokio::fs::rename(
+                    object_dir.join(crate::disk::STORAGE_FORMAT_FILE),
+                    backup.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP),
+                )
+                .await
+                .expect("retain the actual valid metadata as a historical backup");
+                let backup = backup.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP);
+                let bytes = tokio::fs::read(&backup).await.expect("read the historical backup");
+                backups.push((backup, bytes));
+            }
+        }
 
-        let truncated = ListObjectsInfo {
-            is_truncated: true,
-            ..Default::default()
-        };
-        assert!(!should_purge_empty_directory_listing("ghost/", None, 1, false, &truncated));
-    }
+        let delimited = store
+            .clone()
+            .list_objects_generic(bucket, "", None, Some("/".to_owned()), 1000, false)
+            .await
+            .expect("delimiter listing should succeed");
+        assert!(delimited.objects.is_empty());
+        assert!(delimited.prefixes.is_empty(), "residue must not appear as a common prefix");
 
-    #[test]
-    fn empty_directory_listing_does_not_purge_rejected_leading_slash_prefixes() {
-        let empty = ListObjectsInfo::default();
-        for prefix in ["/", "//", "/ghost/"] {
+        let result = store
+            .clone()
+            .list_objects_generic(bucket, "", None, None, 1000, false)
+            .await
+            .expect("recursive bucket listing should succeed");
+        assert!(result.objects.is_empty());
+        let exact_prefix = store
+            .clone()
+            .list_objects_generic(bucket, "repro/", None, None, 1000, false)
+            .await
+            .expect("recursive prefix listing should succeed");
+        assert!(exact_prefix.objects.is_empty());
+        for (path, bytes) in backups {
+            assert_eq!(tokio::fs::read(path).await.expect("LIST must preserve backup"), bytes);
+        }
+        for dir in &dirs {
             assert!(
-                !should_purge_empty_directory_listing(prefix, None, 1000, false, &empty),
-                "rejected prefix {prefix:?} must not initiate orphan cleanup"
+                dir.path().join(bucket).join("repro").exists(),
+                "LIST must preserve the historical residue"
             );
         }
-        assert!(should_purge_empty_directory_listing("ghost/", None, 1000, false, &empty));
-    }
-
-    #[test]
-    fn recursive_bucket_orphan_purge_requires_an_empty_complete_root_scan() {
-        let empty = ListObjectsInfo::default();
-        assert!(super::should_purge_empty_recursive_bucket_listing("", None, None, 1, false, &empty));
-        assert!(!super::should_purge_empty_recursive_bucket_listing(
-            "ghost/", None, None, 1, false, &empty
-        ));
-        assert!(!super::should_purge_empty_recursive_bucket_listing("", Some("/"), None, 1, false, &empty));
-        assert!(!super::should_purge_empty_recursive_bucket_listing(
-            "",
-            None,
-            Some("marker"),
-            1,
-            false,
-            &empty
-        ));
-        assert!(!super::should_purge_empty_recursive_bucket_listing("", None, None, 0, false, &empty));
-        assert!(!super::should_purge_empty_recursive_bucket_listing("", None, None, 1, true, &empty));
-
-        let live = ListObjectsInfo {
-            objects: vec![ObjectInfo::default()],
-            ..Default::default()
-        };
-        assert!(!super::should_purge_empty_recursive_bucket_listing("", None, None, 1, false, &live));
-
-        let incomplete = ListObjectsInfo {
-            is_truncated: true,
-            ..Default::default()
-        };
-        assert!(!super::should_purge_empty_recursive_bucket_listing("", None, None, 1, false, &incomplete));
+        for object in ["repro/obj-00001.bin", "repro/obj-00002.bin"] {
+            store
+                .delete_object(bucket, object, crate::object_api::ObjectOptions::default())
+                .await
+                .expect("fresh exact-key DELETE should reclaim eligible legacy residue");
+            for dir in &dirs {
+                assert!(!dir.path().join(bucket).join(object).exists(), "exact DELETE must reclaim the root");
+            }
+        }
+        store
+            .delete_bucket(bucket, &DeleteBucketOptions::default())
+            .await
+            .expect("the bucket is empty once the residue is reclaimed");
     }
 
     #[tokio::test]
-    async fn empty_recursive_listing_purges_committed_delete_residue() {
+    async fn empty_recursive_listing_preserves_committed_delete_residue() {
         use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
         use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
 
@@ -5062,20 +5029,73 @@ mod test {
         assert!(result.objects.is_empty());
         assert!(result.prefixes.is_empty());
         for dir in &dirs {
+            let residue = dir.path().join(bucket).join("ghost/nested/object").join(data_dir.to_string());
+            assert_eq!(tokio::fs::read(residue.join("part.1")).await.expect("LIST preserves part"), b"stale");
             assert!(
-                !dir.path().join(bucket).join("ghost").exists(),
-                "the empty listing should reclaim its committed delete residue"
+                residue
+                    .join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction))
+                    .exists()
             );
         }
     }
 
-    // #6898: what an abandoned DeleteObjects left on every disk (rollback
-    // backup plus marked data dir) no longer blocks DeleteBucket after an empty
-    // recursive listing of the bucket.
     #[tokio::test]
-    async fn empty_recursive_listing_purges_stranded_delete_rollback_backups() {
+    async fn large_preserved_residue_does_not_create_a_common_prefix() {
+        use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let (dirs, store) = isolated_store_over_temp_disks().await;
+        let bucket = "large-listing-residue";
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        for object in 0..20 {
+            let data_dir = uuid::Uuid::new_v4();
+            let transaction = uuid::Uuid::new_v4();
+            for dir in &dirs {
+                let residue = dir
+                    .path()
+                    .join(bucket)
+                    .join(format!("ghost/{object:02}"))
+                    .join(data_dir.to_string());
+                tokio::fs::create_dir_all(&residue).await.expect("create residue");
+                tokio::fs::write(residue.join("part.1"), b"stale")
+                    .await
+                    .expect("write residue part");
+                tokio::fs::write(
+                    residue.join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction)),
+                    [],
+                )
+                .await
+                .expect("write residue marker");
+            }
+        }
+
+        let recursive = store
+            .clone()
+            .list_objects_generic(bucket, "", None, None, 1000, false)
+            .await
+            .expect("list recursively");
+        assert!(recursive.objects.is_empty());
+        let delimited = store
+            .list_objects_generic(bucket, "", None, Some("/".to_owned()), 1000, false)
+            .await
+            .expect("list with delimiter");
+        assert!(delimited.objects.is_empty());
+        assert!(delimited.prefixes.is_empty(), "residue must not produce a common prefix");
+        for dir in &dirs {
+            assert!(dir.path().join(bucket).join("ghost/00").exists(), "listing must not remove residue");
+        }
+    }
+
+    // A corrupt historical backup must survive LIST and a failed exact DELETE.
+    #[tokio::test]
+    async fn empty_recursive_listing_preserves_corrupt_delete_rollback_backups() {
         use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
         use crate::storage_api_contracts::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
+        use crate::storage_api_contracts::object::ObjectOperations as _;
 
         let (dirs, store) = isolated_store_over_temp_disks().await;
         let bucket = "listing-purge-rollback-backup";
@@ -5119,21 +5139,38 @@ mod test {
             .expect("recursive bucket listing should succeed");
         assert!(result.objects.is_empty());
         for dir in &dirs {
-            assert!(
-                !dir.path().join(bucket).join("repro").exists(),
-                "the empty listing should reclaim the stranded delete residue"
+            let backup = dir
+                .path()
+                .join(bucket)
+                .join("repro/obj-00001.bin")
+                .join(transaction.to_string())
+                .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP);
+            assert_eq!(tokio::fs::read(backup).await.expect("LIST preserves backup"), b"rollback metadata");
+        }
+        let err = store
+            .delete_object(bucket, "repro/obj-00001.bin", crate::object_api::ObjectOptions::default())
+            .await
+            .expect_err("corrupt metadata must not be reclaimed");
+        assert!(matches!(err, StorageError::FileCorrupt));
+        assert!(store.delete_bucket(bucket, &DeleteBucketOptions::default()).await.is_err());
+        for dir in &dirs {
+            let backup = dir
+                .path()
+                .join(bucket)
+                .join("repro/obj-00001.bin")
+                .join(transaction.to_string())
+                .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP);
+            assert_eq!(
+                tokio::fs::read(backup).await.expect("failed cleanup preserves backup"),
+                b"rollback metadata"
             );
         }
-        store
-            .delete_bucket(bucket, &DeleteBucketOptions::default())
-            .await
-            .expect("the bucket is empty once the residue is reclaimed");
     }
 
     #[tokio::test]
-    async fn empty_recursive_bucket_listing_purges_orphan_directory_prefixes() {
+    async fn empty_recursive_bucket_listing_preserves_empty_directory_prefixes() {
         use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
-        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
 
         let (dirs, store) = isolated_store_over_temp_disks().await;
         let bucket = "recursive-bucket-orphan-purge";
@@ -5159,10 +5196,14 @@ mod test {
         assert!(result.prefixes.is_empty(), "a delimiter-less listing has no CommonPrefixes");
         for dir in &dirs {
             assert!(
-                !dir.path().join(bucket).join("ghost").exists(),
-                "an empty recursive bucket scan should reclaim its metadata-less orphan prefix"
+                dir.path().join(bucket).join("ghost/nested/leaf").exists(),
+                "LIST must preserve empty directories"
             );
         }
+        store
+            .delete_bucket(bucket, &DeleteBucketOptions::default())
+            .await
+            .expect("bucket deletion removes empty directories");
 
         let versioned_bucket = "recursive-bucket-versioned-orphan";
         store
@@ -5314,7 +5355,7 @@ mod test {
     }
 
     #[tokio::test]
-    async fn empty_delimiter_listing_hides_and_purges_committed_delete_residue() {
+    async fn empty_delimiter_listing_hides_and_preserves_committed_delete_residue() {
         use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
         use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
 
@@ -5359,15 +5400,22 @@ mod test {
         assert!(result.objects.is_empty());
         assert!(result.prefixes.is_empty(), "delete residue must not surface as a prefix");
         for dir in &dirs {
+            let residue = dir
+                .path()
+                .join(bucket)
+                .join("metrics/2026/object.parquet")
+                .join(data_dir.to_string());
+            assert_eq!(tokio::fs::read(residue.join("part.1")).await.expect("LIST preserves part"), b"stale");
             assert!(
-                !dir.path().join(bucket).join("metrics").join("2026").exists(),
-                "the empty delimiter listing should reclaim the committed delete residue under it"
+                residue
+                    .join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction))
+                    .exists()
             );
         }
     }
 
     #[tokio::test]
-    async fn empty_delimiter_listing_of_residue_ancestor_hides_and_purges_whole_tree() {
+    async fn empty_delimiter_listing_of_residue_ancestor_hides_and_preserves_whole_tree() {
         use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
         use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
 
@@ -5400,9 +5448,7 @@ mod test {
             .expect("committed delete marker should be written");
         }
 
-        // Browsing an ancestor of the deleted key must not show the empty
-        // date folders, and the empty result reclaims the whole residue tree
-        // in one pass instead of one level per listing.
+        // Browsing an ancestor must not expose the deleted key's date folders.
         let result = store
             .clone()
             .list_objects_generic(bucket, "metrics/", None, Some("/".to_owned()), 1000, false)
@@ -5411,15 +5457,22 @@ mod test {
         assert!(result.objects.is_empty());
         assert!(result.prefixes.is_empty(), "ancestors of delete residue must not surface as prefixes");
         for dir in &dirs {
+            let residue = dir
+                .path()
+                .join(bucket)
+                .join("metrics/kubelet/2026/08/28/23/74992556388248657933757.parquet")
+                .join(data_dir.to_string());
+            assert_eq!(tokio::fs::read(residue.join("part.1")).await.expect("LIST preserves part"), b"stale");
             assert!(
-                !dir.path().join(bucket).join("metrics").exists(),
-                "the empty delimiter listing should reclaim the committed delete residue tree under it"
+                residue
+                    .join(format!("{}{}", crate::disk::local::DELETE_DATA_DIR_MARKER_PREFIX, transaction))
+                    .exists()
             );
         }
     }
 
     #[tokio::test]
-    async fn empty_delimiter_listing_of_mixed_residue_ancestor_reclaims_only_committed_subtree() {
+    async fn empty_delimiter_listing_of_mixed_residue_ancestor_preserves_every_subtree() {
         use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
         use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
 
@@ -5492,9 +5545,18 @@ mod test {
         assert!(result.prefixes.is_empty(), "residue subtrees may not surface as prefixes");
         for dir in &dirs {
             let metrics = dir.path().join(bucket).join("metrics");
-            assert!(
-                !metrics.join("cpu").exists(),
-                "the committed subtree must be reclaimed even though a sibling subtree is blocked"
+            let committed = metrics.join("cpu/2026/08/28/22/a.parquet").join(committed_dir.to_string());
+            assert_eq!(
+                tokio::fs::read(committed.join("part.1"))
+                    .await
+                    .expect("LIST preserves committed part"),
+                b"stale"
+            );
+            assert_eq!(
+                tokio::fs::read(committed.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP))
+                    .await
+                    .expect("LIST preserves backup"),
+                b"rollback metadata"
             );
             assert!(
                 metrics
@@ -5502,7 +5564,7 @@ mod test {
                     .join(unmarked_dir.to_string())
                     .join("part.1")
                     .exists(),
-                "residue without a committed marker must survive the purge"
+                "residue without a committed marker must survive the listing"
             );
             assert_eq!(
                 tokio::fs::read(
@@ -6419,6 +6481,147 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    async fn latest_get_follows_read_quorum_on_torn_writes() {
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+        use crate::storage_api_contracts::object::ObjectOperations as _;
+
+        // HEAD and GET serve the version with the most copies once it reaches
+        // the object's data blocks, and a tie goes to the newer version, as in
+        // MinIO's getObjectFileInfo and as heal converges the set. GET used to
+        // demand write quorum whenever enough metadata answered, so a 2+2 or
+        // 8+8 split, or a newer side that could still hold write quorum
+        // through an offline drive, listed but answered 503.
+        let opts = ObjectOptions {
+            suppress_read_repair: true,
+            ..Default::default()
+        };
+        for (drives, parity) in [(4_usize, 2_usize), (16, 8), (16, 4)] {
+            let data = drives - parity;
+            let bucket = format!("torn-write-get-{drives}-{parity}");
+            let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks(drives, parity).await;
+            set.make_bucket(&bucket, &MakeBucketOptions::default())
+                .await
+                .expect("bucket should be created");
+            // Two committed generations of one key give every drive its own
+            // inline shard of each generation; the cases copy them per drive.
+            let mut generations = Vec::new();
+            for body in [b"old generation".as_slice(), b"new generation".as_slice()] {
+                let info = set
+                    .put_object(
+                        &bucket,
+                        "template",
+                        &mut PutObjReader::from_vec(body.to_vec()),
+                        &ObjectOptions {
+                            no_lock: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("generation PUT should commit on every drive");
+                let mut metadata = Vec::with_capacity(drives);
+                for dir in &dirs {
+                    metadata.push(
+                        tokio::fs::read(dir.path().join(&bucket).join("template").join(STORAGE_FORMAT_FILE))
+                            .await
+                            .expect("generation metadata should carry the inline shard"),
+                    );
+                }
+                generations.push((body, info.etag.expect("PUT should report an etag"), metadata));
+            }
+            let (new_body, new_etag, new) = generations.pop().expect("new generation");
+            let (old_body, old_etag, old) = generations.pop().expect("old generation");
+
+            // New copies occupy the first drives, so the offline drive always
+            // removes one new copy. Each phase reads its own keys so the GET
+            // metadata cache never answers for the other phase.
+            let mut cases = Vec::new();
+            for offline in [false, true] {
+                for copies in 1..drives {
+                    for overwrite in [false, true] {
+                        let name = format!(
+                            "{}-{}-{copies:02}",
+                            if offline { "offline" } else { "online" },
+                            if overwrite { "overwrite" } else { "new" }
+                        );
+                        for slot in 0..drives {
+                            let bytes = if slot < copies {
+                                &new[slot]
+                            } else if overwrite {
+                                &old[slot]
+                            } else {
+                                continue;
+                            };
+                            let path = dirs[slot].path().join(&bucket).join(&name);
+                            tokio::fs::create_dir_all(&path).await.expect("create object directory");
+                            tokio::fs::write(path.join(STORAGE_FORMAT_FILE), bytes)
+                                .await
+                                .expect("write generation metadata");
+                        }
+                        cases.push((offline, name, copies, overwrite));
+                    }
+                }
+            }
+
+            for offline in [false, true] {
+                let parked = if offline { set.disks.write().await[0].take() } else { None };
+                for (_, name, copies, overwrite) in cases.iter().filter(|case| case.0 == offline) {
+                    let new_copies = copies - usize::from(offline);
+                    let old_copies = if *overwrite { drives - copies } else { 0 };
+                    let expected = if new_copies >= data && new_copies >= old_copies {
+                        Some((&new_etag, new_body))
+                    } else if old_copies >= data {
+                        Some((&old_etag, old_body))
+                    } else {
+                        None
+                    };
+                    let context = format!("{drives} drives EC {data}+{parity}, {name}: {new_copies} new, {old_copies} old");
+                    let head = set.get_object_info(&bucket, name, &opts).await;
+                    let get = set.get_object_reader(&bucket, name, None, HeaderMap::new(), &opts).await;
+                    match (expected, head, get) {
+                        (Some((etag, body)), Ok(info), Ok(mut reader)) => {
+                            assert_eq!(info.etag.as_ref(), Some(etag), "{context}: HEAD");
+                            assert_eq!(reader.object_info.etag.as_ref(), Some(etag), "{context}: GET");
+                            let mut streamed = Vec::new();
+                            reader
+                                .stream
+                                .read_to_end(&mut streamed)
+                                .await
+                                .unwrap_or_else(|err| panic!("{context}: GET body: {err:?}"));
+                            assert_eq!(streamed, body, "{context}: GET body");
+                        }
+                        (Some((etag, _)), head, get) => panic!(
+                            "{context}: expected {etag}, HEAD {:?}, GET {:?}",
+                            head.map(|info| info.etag),
+                            get.map(|reader| reader.object_info.etag)
+                        ),
+                        (None, head, get) => {
+                            for (method, result) in [
+                                ("HEAD", head.map(|info| info.etag)),
+                                ("GET", get.map(|reader| reader.object_info.etag)),
+                            ] {
+                                let err = result.expect_err(&format!("{context}: {method} expected 404 or unavailable"));
+                                assert!(
+                                    matches!(
+                                        err,
+                                        StorageError::ObjectNotFound(..)
+                                            | StorageError::FileNotFound
+                                            | StorageError::ErasureReadQuorum
+                                            | StorageError::InsufficientReadQuorum(..)
+                                    ),
+                                    "{context}: {method} expected 404 or unavailable, got {err:?}"
+                                );
+                            }
+                        }
+                    }
+                }
+                if offline {
+                    set.disks.write().await[0] = parked;
+                }
+            }
+        }
+    }
+
     #[tokio::test]
     async fn latest_listing_and_get_agree_on_hermetic_set_states() {
         #[derive(Clone, Copy)]
@@ -6491,6 +6694,10 @@ mod test {
             ("d-torn-two", [Some(Generation::New), Some(Generation::New), None, None]),
             ("e-torn-one", [Some(Generation::New), None, None, None]),
             ("f-stale-one", [None, None, None, Some(Generation::Old)]),
+            (
+                "fa-overwrite-two-of-three",
+                [Some(Generation::New), Some(Generation::New), Some(Generation::Old), None],
+            ),
             (
                 "g-overwrite-torn-one",
                 [
@@ -6535,6 +6742,7 @@ mod test {
             ("b-copied", &new_etag),
             ("c-split", &new_etag),
             ("d-torn-two", &new_etag),
+            ("fa-overwrite-two-of-three", &new_etag),
             ("g-overwrite-torn-one", &old_etag),
         ]
         .into_iter()
@@ -6561,12 +6769,6 @@ mod test {
                         "{name}: GET must serve the version LIST selected"
                     );
                 }
-                // Known divergence: when every drive answers, GET of the latest
-                // version demands write quorum (three of four), so a 2+2 split
-                // answers 503 although LIST shows the newer generation, as
-                // MinIO's listing does. F.0's dangling classification does not
-                // cover a split; a GET change is a separate decision.
-                (Some(_), Err(StorageError::InsufficientReadQuorum(..))) if name == "c-split" => {}
                 (Some(_), Err(err)) => assert!(
                     matches!(
                         err,

crates/ecstore/src/store/list_objects/differential_tests.rs

@@ -755,7 +755,7 @@ async fn list_objects_shared_corpus_versions_delete_markers_and_free_versions()
 }
 
 #[tokio::test]
-async fn list_objects_shared_corpus_orphan_cleanup_requires_authoritative_evidence() {
+async fn list_objects_shared_corpus_orphan_listing_preserves_inputs() {
     for authoritative in [false, true] {
         let fixture = Fixture::new(
             &format!("list-corpus-orphan-{authoritative}"),
@@ -774,10 +774,9 @@ async fn list_objects_shared_corpus_orphan_cleanup_requires_authoritative_eviden
         let result = fixture.page("ghost/", None, None, 1).await;
         assert!(result.objects.is_empty() && result.prefixes.is_empty() && !result.is_truncated);
         for dir in &fixture.dirs {
-            assert_eq!(
-                dir.path().join(&fixture.bucket).join("ghost").exists(),
-                !authoritative,
-                "unknown metadata must fail closed"
+            assert!(
+                dir.path().join(&fixture.bucket).join("ghost/nested/leaf").exists(),
+                "LIST must preserve orphan input"
             );
             assert!(
                 dir.path()
@@ -785,7 +784,7 @@ async fn list_objects_shared_corpus_orphan_cleanup_requires_authoritative_eviden
                     .join("legal/child")
                     .join(STORAGE_FORMAT_FILE)
                     .exists(),
-                "purge must preserve legal siblings"
+                "LIST must preserve legal siblings"
             );
         }
         fixture.assert_walk("legal/", None, None, 1).await;
@@ -796,7 +795,7 @@ async fn list_objects_shared_corpus_orphan_cleanup_requires_authoritative_eviden
                     .join("legal/child")
                     .join(STORAGE_FORMAT_FILE)
                     .exists(),
-                "an orphan in the same subtree must not cause the legitimate child to be purged"
+                "an orphan in the same subtree must not affect a legitimate child"
             );
         }
         if authoritative {
@@ -811,17 +810,15 @@ async fn list_objects_shared_corpus_orphan_cleanup_requires_authoritative_eviden
                 .bucket_metadata_sys()
                 .expect("authoritative fixture has a metadata system");
             let guard = metadata_sys.write().await;
-            assert!(
-                timeout(Duration::from_millis(100), fixture.page("timed-out/", None, None, 1))
-                    .await
-                    .is_err(),
-                "blocked authoritative evidence must time out before cleanup can classify the prefix as empty"
-            );
+            let result = timeout(Duration::from_secs(2), fixture.page("timed-out/", None, None, 1))
+                .await
+                .expect("LIST must not wait for cleanup authorization");
+            assert!(result.objects.is_empty() && result.prefixes.is_empty() && !result.is_truncated);
             drop(guard);
             for dir in &fixture.dirs {
                 assert!(
                     dir.path().join(&fixture.bucket).join("timed-out").exists(),
-                    "a timed-out evidence check must preserve its orphan input"
+                    "LIST must preserve its orphan input while metadata is locked"
                 );
             }
         }

crates/ecstore/src/store/object.rs

@@ -5139,27 +5139,6 @@ impl ECStore {
         purged
     }
 
-    pub(super) async fn purge_orphan_dir_objects_in_bucket(&self, bucket: &str) -> bool {
-        if is_meta_bucketname(bucket) {
-            return false;
-        }
-        for pool in self.pools.iter() {
-            for set in pool.disk_set.iter() {
-                if !set.orphan_purge_has_complete_disk_set().await {
-                    return false;
-                }
-            }
-        }
-
-        let mut purged = false;
-        for pool in self.pools.iter() {
-            for set in pool.disk_set.iter() {
-                purged |= set.purge_orphan_dir_objects_in_bucket(bucket).await;
-            }
-        }
-        purged
-    }
-
     /// Delete one object (or a prefix) on behalf of a caller that may be
     /// dropped mid-way, such as an S3 request whose client disconnects.
     ///

crates/io-metrics/benches/metrics_pipeline.rs

@@ -1,67 +1,10 @@
 use criterion::{Criterion, criterion_group, criterion_main};
-use metrics_util::debugging::DebuggingRecorder;
-use rustfs_io_metrics::{
-    LocalBatchReadDecision, LocalBatchReadMetrics, LocalBatchReadStage, record_local_batch_read_decision,
-    set_get_stage_metrics_enabled,
-};
 use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics, record_get_object_timeout_operation_start};
 use rustfs_io_metrics::{record_s3_op, s3_http_metrics::S3HttpRequestGuard};
 use rustfs_s3_ops::S3Operation;
 use std::hint::black_box;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
-
-fn local_batch_attribution_sample() {
-    let prepare = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now);
-    record_local_batch_read_decision(LocalBatchReadDecision::Selected, prepare.map(|started| started.elapsed()));
-    if let Some(metrics) = LocalBatchReadMetrics::capture() {
-        metrics.record_batch(4);
-        let execution = Instant::now();
-        metrics.record_stage(LocalBatchReadStage::QueueWait, true, execution.elapsed());
-        for _ in 0..4 {
-            for stage in [
-                LocalBatchReadStage::Metadata,
-                LocalBatchReadStage::Open,
-                LocalBatchReadStage::Read,
-            ] {
-                let started = Instant::now();
-                metrics.record_stage(stage, true, started.elapsed());
-            }
-            metrics.record_shard(true, 32 * 1024, 32 * 1024);
-        }
-        metrics.record_stage(LocalBatchReadStage::Execution, true, execution.elapsed());
-    }
-}
-
-fn bench_local_batch_attribution(c: &mut Criterion) {
-    let mut group = c.benchmark_group("local_batch_attribution_4_shards_debug_recorder");
-    for (name, enabled) in [("disabled", false), ("enabled", true)] {
-        set_get_stage_metrics_enabled(enabled);
-        group.bench_function(name, |b| {
-            b.iter_custom(|iterations| {
-                let mut remaining = iterations;
-                let mut elapsed = Duration::ZERO;
-                while remaining > 0 {
-                    // Bound captured histogram storage independently of Criterion's
-                    // iteration count. Recorder setup/teardown is not measured.
-                    let chunk = remaining.min(128);
-                    let recorder = DebuggingRecorder::new();
-                    metrics::with_local_recorder(&recorder, || {
-                        let started = Instant::now();
-                        for _ in 0..chunk {
-                            local_batch_attribution_sample();
-                        }
-                        elapsed += started.elapsed();
-                    });
-                    remaining -= chunk;
-                }
-                elapsed
-            });
-        });
-    }
-    group.finish();
-    set_get_stage_metrics_enabled(false);
-}
+use std::time::Duration;
 
 fn bench_record_get_object_timeout_operation_start(c: &mut Criterion) {
     c.bench_function("record_get_object_timeout_operation_start", |b| {
@@ -111,7 +54,6 @@ criterion_group!(
     bench_record_get_object_timeout_operation_start,
     bench_s3_http_outcomes,
     bench_update_concurrent_requests,
-    bench_metrics_collector_record_io_operation,
-    bench_local_batch_attribution
+    bench_metrics_collector_record_io_operation
 );
 criterion_main!(benches);

crates/io-metrics/src/lib.rs

@@ -49,11 +49,6 @@
 #[macro_use]
 extern crate metrics;
 
-mod local_batch_read;
-pub use local_batch_read::{
-    LocalBatchReadDecision, LocalBatchReadMetrics, LocalBatchReadStage, record_local_batch_read_decision,
-};
-
 use std::collections::HashMap;
 use std::sync::{
     LazyLock, Mutex,

crates/io-metrics/src/local_batch_read.rs

@@ -1,210 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Opt-in attribution for the buffered local shard batch path.
-
-use crate::get_stage_metrics_enabled;
-use metrics::{Counter, Histogram};
-use std::time::Duration;
-
-/// One decision per reader-setup attempt, rather than per S3 request.
-#[derive(Clone, Copy, Debug)]
-pub enum LocalBatchReadDecision {
-    Selected,
-    MmapDisabled,
-    CopySource,
-    ReadCapExceeded,
-    InlineData,
-    NonLocalShard,
-    InvalidPath,
-    NoLocalShards,
-    UnsupportedPlatform,
-}
-
-impl LocalBatchReadDecision {
-    const fn as_str(self) -> &'static str {
-        match self {
-            Self::Selected => "selected",
-            Self::MmapDisabled => "mmap_disabled",
-            Self::CopySource => "copy_source",
-            Self::ReadCapExceeded => "read_cap_exceeded",
-            Self::InlineData => "inline_data",
-            Self::NonLocalShard => "non_local_shard",
-            Self::InvalidPath => "invalid_path",
-            Self::NoLocalShards => "no_local_shards",
-            Self::UnsupportedPlatform => "unsupported_platform",
-        }
-    }
-}
-
-/// Record eligibility, including path preparation only when it actually ran.
-pub fn record_local_batch_read_decision(decision: LocalBatchReadDecision, preparation: Option<Duration>) {
-    if !get_stage_metrics_enabled() {
-        return;
-    }
-    let decision = decision.as_str();
-    counter!("rustfs_io_get_object_local_batch_decisions_total", "decision" => decision).increment(1);
-    if let Some(duration) = preparation {
-        histogram!("rustfs_io_get_object_local_batch_prepare_seconds", "decision" => decision).record(duration.as_secs_f64());
-    }
-}
-
-/// Stages of the existing buffered positioned-read implementation.
-#[derive(Clone, Copy)]
-pub enum LocalBatchReadStage {
-    QueueWait,
-    Execution,
-    Metadata,
-    Open,
-    Read,
-}
-
-impl LocalBatchReadStage {
-    const fn index(self) -> usize {
-        match self {
-            Self::QueueWait => 0,
-            Self::Execution => 1,
-            Self::Metadata => 2,
-            Self::Open => 3,
-            Self::Read => 4,
-        }
-    }
-}
-
-/// Handles captured before dispatch, kept alive by the actual blocking worker.
-///
-/// Capturing the recorder once avoids per-I/O registration and also preserves
-/// completion samples when the caller stops waiting. No timers or handles are
-/// created with GET-stage metrics disabled. The snapshot applies to the whole
-/// submitted batch, including work that finishes after the switch is disabled.
-pub struct LocalBatchReadMetrics {
-    stages: [[Histogram; 2]; 5],
-    shards: [Counter; 2],
-    requested_bytes: [Counter; 2],
-    read_bytes: [Counter; 2],
-    batch_size: Histogram,
-}
-
-fn outcome(success: bool) -> &'static str {
-    if success { "success" } else { "error" }
-}
-
-impl LocalBatchReadMetrics {
-    /// Resolve bounded-label handles only for an explicitly instrumented batch.
-    /// Boxing the enabled state keeps the blocking task small when disabled.
-    pub fn capture() -> Option<Box<Self>> {
-        if !get_stage_metrics_enabled() {
-            return None;
-        }
-        let stages = ["queue_wait", "execution", "metadata", "open", "read"].map(|stage| {
-            [false, true].map(|success| {
-                histogram!("rustfs_io_get_object_local_batch_stage_seconds", "stage" => stage, "outcome" => outcome(success))
-            })
-        });
-        Some(Box::new(Self {
-            stages,
-            shards: [false, true]
-                .map(|success| counter!("rustfs_io_get_object_local_batch_shards_total", "outcome" => outcome(success))),
-            requested_bytes: [false, true]
-                .map(|success| counter!("rustfs_io_get_object_local_batch_requested_bytes_total", "outcome" => outcome(success))),
-            read_bytes: [false, true]
-                .map(|success| counter!("rustfs_io_get_object_local_batch_read_bytes_total", "outcome" => outcome(success))),
-            batch_size: histogram!("rustfs_io_get_object_local_batch_shards"),
-        }))
-    }
-
-    /// Count shards actually submitted to this blocking batch.
-    pub fn record_batch(&self, shards: usize) {
-        self.batch_size.record(crate::shard_read_fanout_to_f64(shards));
-    }
-
-    /// Record an executed stage; unattempted stages receive no sample.
-    ///
-    /// Queue wait succeeds when the worker starts. Execution covers the whole
-    /// batch, while metadata/open/read describe individual I/O operations.
-    /// Read timing excludes buffer allocation; execution timing includes it.
-    pub fn record_stage(&self, stage: LocalBatchReadStage, success: bool, duration: Duration) {
-        self.stages[stage.index()][usize::from(success)].record(duration.as_secs_f64());
-    }
-
-    /// Separate completed shard outcomes and bytes actually returned by read_at.
-    pub fn record_shard(&self, success: bool, requested: usize, read: usize) {
-        let index = usize::from(success);
-        self.shards[index].increment(1);
-        self.requested_bytes[index].increment(u64::try_from(requested).unwrap_or(u64::MAX));
-        self.read_bytes[index].increment(u64::try_from(read).unwrap_or(u64::MAX));
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::set_get_stage_metrics_enabled;
-    use crate::tests::{METRICS_FLAG_LOCK, counter_total, histogram_samples};
-    use metrics_util::debugging::DebuggingRecorder;
-    use std::time::Duration;
-
-    #[test]
-    fn local_batch_read_metrics_are_disabled_without_stage_opt_in() {
-        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|error| error.into_inner());
-        let recorder = DebuggingRecorder::new();
-        let snapshotter = recorder.snapshotter();
-        metrics::with_local_recorder(&recorder, || {
-            set_get_stage_metrics_enabled(false);
-            assert!(LocalBatchReadMetrics::capture().is_none());
-            record_local_batch_read_decision(LocalBatchReadDecision::Selected, Some(Duration::from_millis(1)));
-        });
-        assert!(snapshotter.snapshot().into_vec().is_empty());
-    }
-
-    #[test]
-    fn local_batch_read_handles_record_worker_costs_and_actual_bytes() {
-        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|error| error.into_inner());
-        let recorder = DebuggingRecorder::new();
-        let snapshotter = recorder.snapshotter();
-        let observation = metrics::with_local_recorder(&recorder, || {
-            set_get_stage_metrics_enabled(true);
-            record_local_batch_read_decision(LocalBatchReadDecision::Selected, Some(Duration::from_millis(2)));
-            LocalBatchReadMetrics::capture().expect("enabled batch metrics must capture recorder handles")
-        });
-        set_get_stage_metrics_enabled(false);
-        std::thread::spawn(move || {
-            observation.record_stage(LocalBatchReadStage::QueueWait, true, Duration::from_millis(3));
-            observation.record_stage(LocalBatchReadStage::Metadata, false, Duration::from_millis(4));
-            observation.record_shard(false, 1024, 0);
-            observation.record_shard(true, 16, 16);
-            observation.record_batch(2);
-        })
-        .join()
-        .expect("worker should record through the captured handles");
-        set_get_stage_metrics_enabled(false);
-
-        let rows = snapshotter.snapshot().into_vec();
-        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_decisions_total"), Some(1));
-        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_shards_total"), Some(2));
-        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_requested_bytes_total"), Some(1040));
-        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_read_bytes_total"), Some(16));
-        assert_eq!(histogram_samples(&rows, "rustfs_io_get_object_local_batch_prepare_seconds"), vec![0.002]);
-        assert_eq!(
-            histogram_samples(&rows, "rustfs_io_get_object_local_batch_stage_seconds"),
-            vec![0.003, 0.004]
-        );
-        assert_eq!(histogram_samples(&rows, "rustfs_io_get_object_local_batch_shards"), vec![2.0]);
-        for (key, _, _, _) in rows {
-            for label in key.key().labels() {
-                assert!(matches!(label.key(), "decision" | "stage" | "outcome"));
-            }
-        }
-    }
-}

crates/obs/src/telemetry/otel.rs

@@ -83,16 +83,6 @@ const GET_OBJECT_DURATION_HISTOGRAM_BUCKETS: &[f64] = &[
     0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
 ];
 
-const LOCAL_BATCH_DURATION_HISTOGRAM_METRICS: &[&str] = &[
-    "rustfs_io_get_object_local_batch_prepare_seconds",
-    "rustfs_io_get_object_local_batch_stage_seconds",
-];
-
-const LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS: &[f64] = &[
-    0.0000001, 0.00000025, 0.0000005, 0.000001, 0.0000025, 0.000005, 0.00001, 0.000025, 0.00005, 0.0001, 0.00025, 0.0005, 0.001,
-    0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
-];
-
 const DEFAULT_OTLP_HTTP_TIMEOUT: Duration = Duration::from_secs(10);
 const OTEL_EXPORTER_OTLP_TIMEOUT: &str = "OTEL_EXPORTER_OTLP_TIMEOUT";
 const OTEL_EXPORTER_OTLP_TRACES_TIMEOUT: &str = "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT";
@@ -491,25 +481,21 @@ fn build_meter_provider(
 }
 
 fn get_object_duration_histogram_view(instrument: &Instrument) -> Option<Stream> {
-    let boundaries = get_object_duration_histogram_buckets(instrument.name())?;
+    if !is_get_object_duration_histogram_metric(instrument.name()) {
+        return None;
+    }
 
     Stream::builder()
         .with_aggregation(Aggregation::ExplicitBucketHistogram {
-            boundaries: boundaries.to_vec(),
+            boundaries: GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.to_vec(),
             record_min_max: true,
         })
         .build()
         .ok()
 }
 
-fn get_object_duration_histogram_buckets(name: &str) -> Option<&'static [f64]> {
-    if LOCAL_BATCH_DURATION_HISTOGRAM_METRICS.contains(&name) {
-        Some(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS)
-    } else if GET_OBJECT_DURATION_HISTOGRAM_METRICS.contains(&name) {
-        Some(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS)
-    } else {
-        None
-    }
+fn is_get_object_duration_histogram_metric(name: &str) -> bool {
+    GET_OBJECT_DURATION_HISTOGRAM_METRICS.contains(&name)
 }
 
 /// Build an optional [`SdkLoggerProvider`] for the given log endpoint.
@@ -1112,22 +1098,10 @@ mod tests {
 
     #[test]
     fn test_get_object_duration_histogram_metric_match_is_scoped() {
-        for name in [
-            "rustfs_io_get_object_stage_duration_seconds",
-            "rustfs_io_get_object_request_duration_seconds",
-        ] {
-            assert_eq!(get_object_duration_histogram_buckets(name), Some(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS));
-        }
-        for name in LOCAL_BATCH_DURATION_HISTOGRAM_METRICS {
-            assert_eq!(get_object_duration_histogram_buckets(name), Some(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS));
-        }
-        for name in [
-            "rustfs_io_put_object_request_duration_seconds",
-            "rustfs_io_get_object_response_size_bytes",
-            "rustfs_io_get_object_local_batch_shards",
-        ] {
-            assert!(get_object_duration_histogram_buckets(name).is_none());
-        }
+        assert!(is_get_object_duration_histogram_metric("rustfs_io_get_object_stage_duration_seconds"));
+        assert!(is_get_object_duration_histogram_metric("rustfs_io_get_object_request_duration_seconds"));
+        assert!(!is_get_object_duration_histogram_metric("rustfs_io_put_object_request_duration_seconds"));
+        assert!(!is_get_object_duration_histogram_metric("rustfs_io_get_object_response_size_bytes"));
     }
 
     #[test]
@@ -1137,17 +1111,6 @@ mod tests {
         assert_eq!(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.last(), Some(&10.0));
     }
 
-    #[test]
-    fn local_batch_duration_buckets_resolve_sub_microsecond_operations() {
-        assert!(
-            LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS
-                .windows(2)
-                .all(|pair| pair[0] < pair[1])
-        );
-        assert_eq!(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS.first(), Some(&0.0000001));
-        assert_eq!(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS.last(), Some(&10.0));
-    }
-
     #[cfg(all(
         feature = "pyroscope",
         any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))

docs/architecture/erasure-coding.md

@@ -313,7 +313,7 @@ Small objects store their payload inline after the container CRC ([filemeta_inli
 ## 8. Read path and read quorum
 
 - **INVARIANT — read quorum = `data_blocks`.** `object_quorum_from_meta` returns `(read_quorum = data_blocks, write_quorum)` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)); `parity_blocks = common_parity(...)` is the parity value held by the most disks that still reaches its own read quorum. When `default_parity_count == 0`, read = write = all shards.
-- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. Latest-version reads may escalate to write_quorum to avoid resurrecting a partially-overwritten version.
+- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. S3 reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set. Reads in the system bucket (`is_meta_bucketname`) keep the commit-proof rule in `select_committed_fileinfo`: when enough copies answer to prove write quorum, the latest version must hold it, and a degraded read never serves a newer version below write quorum. Pool metadata's PREPARE/COMMIT recovery and identity epochs rely on an ordinary read implying a committed record, so a split stays unreadable until native recovery resolves it. When every drive answered and an S3 latest read still cannot select a version, `read_latest_past_dangling_versions` walks the version stack newest first through exact-version reads, as MinIO's `pickLatestQuorumFilesInfo` resolves stacks: it serves the readable version with the newest selected mod_time, skips versions whose exact read proves them dangling (the NoSuchVersion an explicit GET reports) and queues heal for them, and keeps the read-quorum error on any other outcome. A failed versioned PUT that left the new version on fewer drives than its data blocks therefore no longer hides the committed version below it; ListObjectVersions can still show the dangling version as latest until heal removes it. Metadata updates of the latest version (object tagging, `put_object_metadata`) and the CopyObject source read resolve a failed selection through the same walk; when the walk finds no readable version, tagging keeps the read-quorum error while `put_object_metadata` and CopyObject keep their existing `delete_if_dangling` cleanup. Replication status write-back never walks. The update targets only the served version's disks and is refused before writing when they cannot reach its write quorum, including a version-only copy.
 - **INVARIANT — decode needs ≥ `data_blocks` shards.** The stripe reader requires `available_shards ≥ data_shards`; below that the read fails closed with a read-quorum error (never silent truncation) ([set_disk/read.rs](../../crates/ecstore/src/set_disk/read.rs), [set_disk/shard_source.rs](../../crates/ecstore/src/set_disk/shard_source.rs)). Codec geometry is validated at construction: `Erasure::try_new` / `try_new_with_options` ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)) return `ErasureConstructionError` for `data_shards == 0`, `block_size == 0`, shard-count overflow, or an unsupported shard configuration, so a read never reaches a `block_size` / `data_shards` division with invalid geometry (§13).
 - If `available ≥ data_blocks` but some shards are missing, the read is served **and** a background read-repair heal is enqueued.
 - **INVARIANT — cross-stripe read verification.** When a data shard is missing and `available > data_blocks`, reconstruction regenerates parity and compares it to the surviving parity; a mismatch is `InvalidData "inconsistent read source shards"` (backlog#832), catching corruption that passed per-shard bitrot but disagrees across the stripe ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)).

docs/architecture/scanner-usage-publication.md

@@ -325,9 +325,45 @@ worker finishing does not certify its namespace coverage.
 | `bucket_result_rejected` | A bucket result did not match the cycle's frozen tier-registry generation and was not applied. |
 | `set_cache_publish_skipped` | `completed_buckets` counts distinct bucket roots successfully applied to this set cache and `carried_buckets` the partial walks covered by a certified value. Use `first_incomplete_bucket` with pool/set/cycle to locate one actual gap in this set. Inspect `unprocessed_buckets` and budget/cancellation state, then correlate that bucket's worker events. |
 | `set_scan_completed` | Set workers have finished. `coverage_complete` reports the structural coverage gate, not durable publication. This event alone does not prove a complete set snapshot. |
-| `incomplete_coverage` | The cycle did not obtain complete structural coverage. This does not identify a hot bucket or prove that ordinary writes caused the gap. |
+| `incomplete_coverage` | A Partial cycle did not match either of the more specific partial-cause labels. It does not identify which completeness or publication check failed, or prove that a bucket or drive was missed. |
 | `publication_proof_pending` | Publication is waiting for a retryable proof. Use cycle/stage, `pending_duration`, the preceding `publication_proof_started` kind, and the peer/purpose in the error. This event does not assert that the walk was Complete. |
 
+`bucket_sweep_incomplete`, `partial_saved_not_published`,
+`set_cache_publish_skipped`, and `set_scan_completed` are debug events. Their
+absence cannot exclude those paths when debug logging was not collected.
+
+`metrics.last_cycle_bucket_drive_scans` is the local process's change in
+`scan_bucket_drive` timed completions during the cycle. It includes successful,
+partial, namespace-not-found, and non-cancelled error returns; a cancelled
+scan can emit a partial result without incrementing this count. The leader's
+remote progress adds object and directory counts, but not remote bucket-drive
+completions. The count can also include concurrent local scan work. It is not a
+cluster-wide success count or a coverage denominator, so comparing it with
+bucket count multiplied by set or drive count cannot identify missing drives.
+See `Metrics::scan_cycle_work_since` in [scanner metrics](../../crates/scanner-metrics/src/metrics.rs)
+and `ScannerIODisk::nsscanner_disk` in [the disk scanner](../../crates/scanner/src/scanner_io/io_disk.rs).
+
+For a Partial cycle, compare the same leader process's `metrics.usage_candidates`
+in `/rustfs/admin/v3/scanner/status` before and after the cycle, when available.
+The corresponding `rustfs_scanner_usage_candidates_total` counters require an
+enabled exporter; other scanner metrics endpoints can expose a different schema.
+For `kind=completed`, the `outcome` names the first candidate-construction check
+that returned no candidate, or `accepted`; acceptance does not prove durable
+publication. Counters are process-lifetime, not per-cycle or per-bucket, and
+reset on restart. Correlate their deltas with the same cycle and leader epoch in
+logs; concurrent candidate construction and non-atomic sampling can make
+attribution uncertain. An unavailable field or counter is not zero.
+
+The scanner metrics `rustfs_scanner_bucket_drive_result_total` and
+`rustfs_scanner_cycle_bucket_drive_result` expose local results by server, bucket, drive and
+`result` (`success`, `partial`, or `error`); the latter also has `cycle_scope`
+(`current` or `last`). `rustfs_scanner_active_bucket_drive_scans` shows work
+active at the sample time. These are observations, not per-drive terminal acknowledgements:
+the local result table is capped and evicts old keys, the cycle labels do not
+carry a cycle or epoch, and retries can repeat work. A missing row does not
+prove that a terminal was never received. Match observations to the relevant
+node and cycle before comparing them with set-cache and publication logs.
+
 For a report such as 52 completed buckets out of 53, first identify whether
 the missing bucket was Partial, failed, never admitted, or rejected at merge.
 Only then inspect its worker evidence. Mixed coverage is a resumable sweep

docs/operations/delete-transaction-recovery.md

@@ -105,6 +105,12 @@ locking, and movement guards still apply. Prefix recovery settles only matching
 keys, including transactions hidden from listings. Dry-run inspects and reports without
 settling transactions.
 
+LIST does not remove historical residue on updated nodes. During a rolling
+upgrade, older nodes can still remove it while serving LIST. A bucket with
+hidden residue can still fail DeleteBucket; recover known keys with an
+unconditional exact-key DELETE, or inspect and remove eligible residue through
+administrator heal.
+
 An unconditional DELETE of a missing exact key, or an administrator heal with
 `remove` enabled, can authorize historical residue removal only when the
 bucket is authoritatively never-versioned and has no Object Lock. Every

docs/operations/hotpath-warp-ab-runbook.md

@@ -143,39 +143,6 @@ Rules that override the table: a candidate result is actionable only when the `A
 
 ### CPU and memory evidence
 
-#### Local shard batch attribution
-
-The existing GET-stage metric gate also controls local batch attribution. This
-path performs buffered positioned reads, even if O_DIRECT or io_uring is enabled
-for another backend. Keep the gate, exporter, and build features identical
-between compared binaries; calibrate instrumentation overhead separately.
-
-| Metric suffix after `rustfs_io_get_object_local_batch_` | Meaning |
-| --- | --- |
-| `decisions_total` | Reader-setup decisions, labelled by a fixed decision enum; this is not an S3 request counter. |
-| `prepare_seconds` | Eligibility and path preparation, including pre-worker path checks; absent when preparation was skipped. |
-| `stage_seconds` | Queue wait, whole-worker execution, and per-shard metadata/open/read durations, labelled by stage and operation outcome. |
-| `shards` / `shards_total` | Batch size and completed shard outcomes. |
-| `requested_bytes_total` / `read_bytes_total` | Requested bytes and bytes actually returned by positioned reads, split by shard outcome. |
-
-Worker handles are captured before dispatch and remain owned by the worker, so
-completion samples can arrive after the caller stops waiting. A stage that did
-not execute contributes no sample. Queue wait ends when the worker starts;
-execution includes allocation and instrumentation, while read timing excludes
-buffer allocation. These nested and per-shard durations cannot be added to
-request latency, and stage p99 values cannot be summed. Match request-boundary
-samples and successful-request counters to the same workload window when
-normalizing costs. Keep failed requests and partial-shard failures visible.
-Inline GET and HEAD can bypass this reader-setup entry entirely; absence of a
-decision sample does not imply a failed request. The two local duration metrics
-use dedicated 0.1-microsecond-to-10-second histogram boundaries in the OTLP view;
-request-duration histogram boundaries are separate.
-
-For a bounded, four-shard instrumentation microbenchmark, run
-`cargo bench -p rustfs-io-metrics --bench metrics_pipeline -- local_batch_attribution_4_shards_debug_recorder`.
-It performs no disk I/O and uses a debugging recorder, so its result calibrates
-the sampling code rather than production exporter overhead or S3 latency.
-
 Warp output says whether throughput or latency changed; host telemetry says why. Collect it for the whole run and stop the collectors after the script exits.
 
 | Tool | Command | Answers |