fix(ecstore): satisfy clippy in rebalance test

crates/ecstore/src/store/list_objects.rs

@@ -70,7 +70,7 @@ use tokio::io::duplex;
 use tokio::sync::broadcast::{self};
 use tokio::sync::mpsc::{self, Receiver, Sender};
 use tokio::sync::{OnceCell, RwLock};
-use tokio::task::JoinSet;
+use tokio::task::{JoinHandle, JoinSet};
 use tokio_util::sync::CancellationToken;
 use tracing::{Instrument, debug, error, info, warn};
 use uuid::Uuid;
@@ -4331,6 +4331,7 @@ impl ECStore {
             "store list_merged started"
         );
 
+        let rx = rx.child_token();
         let mut futures = Vec::new();
 
         let mut inputs = Vec::new();
@@ -4346,16 +4347,10 @@ impl ECStore {
             }
         }
 
-        tokio::spawn(
-            async move {
-                if let Err(err) = merge_entry_channels(rx, inputs, sender.clone(), 1).await {
-                    error!("merge_entry_channels err {:?}", err)
-                }
-            }
-            .instrument(tracing::Span::current()),
-        );
+        let merge_task = spawn_listing_merge(rx, inputs, sender);
 
         let results = join_all(futures).await;
+        merge_task.await.map_err(Error::from)??;
 
         let mut all_at_eof = true;
 
@@ -4422,6 +4417,7 @@ impl ECStore {
     ) -> Result<()> {
         check_list_objs_args(bucket, prefix, &None)?;
 
+        let rx = rx.child_token();
         let mut futures = Vec::new();
         let mut inputs = Vec::new();
 
@@ -4783,17 +4779,11 @@ impl ECStore {
             .instrument(tracing::Span::current()),
         );
 
-        tokio::spawn(
-            async move {
-                if let Err(err) = merge_entry_channels(rx, inputs, merge_tx, 1).await {
-                    error!("merge_entry_channels err {:?}", err)
-                }
-            }
-            .instrument(tracing::Span::current()),
-        );
+        let merge_task = spawn_listing_merge(rx, inputs, merge_tx);
 
         let walk_started = std::time::Instant::now();
         let walk_results = join_all(futures).await;
+        merge_task.await.map_err(Error::from)??;
         let mut errs = Vec::new();
         for walk_result in walk_results {
             match walk_result {
@@ -5068,6 +5058,106 @@ async fn send_or_cancel(rx: &CancellationToken, out_channel: &Sender<MetaCacheEn
     }
 }
 
+/// Each input has already been resolved inside its own erasure set. This is a
+/// union of version histories, never a quorum vote between unrelated pools.
+fn merge_object_entry_versions(first: &mut MetaCacheEntry, others: impl Iterator<Item = MetaCacheEntry>) -> Result<()> {
+    let name = first.name.clone();
+    let mut versions: HashMap<(Option<Uuid>, bool), (FileMetaShallowVersion, ObjectInfo)> = HashMap::new();
+    for mut entry in std::iter::once(std::mem::take(first)).chain(others) {
+        let meta = match entry.cached.take() {
+            Some(meta) => meta,
+            None => FileMeta::load(&entry.metadata).map_err(|_| Error::FileCorrupt)?,
+        };
+        if meta.versions.is_empty() {
+            return Err(Error::FileCorrupt);
+        }
+        for version in meta.versions {
+            let parsed = version.parse_version_meta().map_err(|_| Error::FileCorrupt)?;
+            if !parsed.valid() || parsed.version_type != version.header.version_type {
+                return Err(Error::FileCorrupt);
+            }
+            let fi = parsed.into_fileinfo("", &name, true).map_err(|_| Error::FileCorrupt)?;
+            let version_id = fi.version_id.filter(|id| !id.is_nil());
+            if version_id != version.header.version_id.filter(|id| !id.is_nil())
+                || fi.mod_time != version.header.mod_time
+                || fi.tier_free_version() != version.header.free_version()
+            {
+                return Err(Error::FileCorrupt);
+            }
+            let info = ObjectInfo::from_file_info(&fi, "", &name, true);
+            let identity = (version_id, version.header.free_version());
+            match versions.entry(identity) {
+                std::collections::hash_map::Entry::Vacant(slot) => {
+                    slot.insert((version, info));
+                }
+                std::collections::hash_map::Entry::Occupied(mut slot) => {
+                    let (previous, previous_info) = slot.get();
+                    // Suspended and unversioned writes replace the one null
+                    // slot. Distinct UUID versions never supersede each other.
+                    if version_id.is_none() && info.mod_time != previous_info.mod_time {
+                        if info.mod_time > previous_info.mod_time {
+                            slot.insert((version, info));
+                        }
+                        continue;
+                    }
+                    let equivalent = if info.delete_marker && previous_info.delete_marker {
+                        super::object::is_equivalent_data_movement_delete_marker(&info, previous_info)
+                    } else {
+                        crate::data_movement::is_equivalent_data_movement_object_identity(&info, previous_info, true, true)
+                    };
+                    if !equivalent {
+                        return Err(Error::FileCorrupt);
+                    }
+                    // Equivalent migrated copies can have different coding or
+                    // data directories. Choose a stable representation without
+                    // making input order part of the S3 version order.
+                    if version.meta < previous.meta {
+                        slot.insert((version, info));
+                    }
+                }
+            }
+        }
+    }
+    let mut merged = FileMeta::new();
+    merged.versions = versions.into_values().map(|(version, _)| version).collect();
+    merged.versions.sort_by(|a, b| {
+        if a.header.sorts_before(&b.header) {
+            std::cmp::Ordering::Less
+        } else if b.header.sorts_before(&a.header) {
+            std::cmp::Ordering::Greater
+        } else {
+            std::cmp::Ordering::Equal
+        }
+    });
+    let metadata = merged.marshal_msg()?;
+    *first = MetaCacheEntry {
+        name,
+        metadata,
+        cached: Some(merged),
+        reusable: true,
+    };
+    Ok(())
+}
+
+/// `rx` is private to the producers. Cancelling it on a merge error must not
+/// cancel the request token, which would suppress that error at the API edge.
+fn spawn_listing_merge(
+    rx: CancellationToken,
+    inputs: Vec<Receiver<MetaCacheEntry>>,
+    sender: Sender<MetaCacheEntry>,
+) -> JoinHandle<Result<()>> {
+    tokio::spawn(
+        async move {
+            let result = merge_entry_channels(rx.clone(), inputs, sender, 1).await;
+            if result.is_err() {
+                rx.cancel();
+            }
+            result
+        }
+        .instrument(tracing::Span::current()),
+    )
+}
+
 async fn merge_entry_channels(
     rx: CancellationToken,
     in_channels: Vec<Receiver<MetaCacheEntry>>,
@@ -5133,6 +5223,7 @@ async fn merge_entry_channels(
     // after anything greater has been emitted).
     let mut last_emitted = String::new();
     let mut group: Vec<Box<MergeHead>> = Vec::new();
+    let mut object_entries: Vec<MetaCacheEntry> = Vec::new();
     let mut refill: Vec<usize> = Vec::with_capacity(in_channels.len());
 
     while let Some(Reverse(first)) = heap.pop() {
@@ -5150,7 +5241,7 @@ async fn merge_entry_channels(
         // Resolve the same-name group to one winner (heads arrive in ascending
         // channel order):
         //  - prefix dir vs prefix dir: the first (lowest channel) wins;
-        //  - object vs object: the later channel wins (legacy authority rule);
+        //  - object vs object: merge the independently resolved version stacks;
         //  - object vs prefix dir: same-name means both end with the separator,
         //    i.e. the object is an explicit "directory marker" for the same S3
         //    key — it shadows the prefix dir so the key does not surface as
@@ -5168,11 +5259,27 @@ async fn merge_entry_channels(
                 if dir_winner.is_none() {
                     dir_winner = Some(head);
                 }
+            } else if let Some(winner) = object_winner.as_ref() {
+                // Key-only candidates carry no version metadata and cannot
+                // replace a resolved stack or contribute a quorum vote.
+                if head.entry.is_object() {
+                    if winner.entry.is_object() {
+                        object_entries.push(head.entry);
+                    } else {
+                        object_winner = Some(head);
+                    }
+                }
             } else {
                 object_winner = Some(head);
             }
         }
 
+        if !object_entries.is_empty()
+            && let Some(winner) = object_winner.as_mut()
+        {
+            merge_object_entry_versions(&mut winner.entry, object_entries.drain(..))?;
+        }
+
         if let Some(head) = object_winner.or(dir_winner)
             && head.entry.name != last_emitted
         {
@@ -5605,6 +5712,7 @@ impl Sets {
             "sets list_merged started"
         );
 
+        let rx = rx.child_token();
         let mut futures = Vec::new();
         let mut inputs = Vec::new();
 
@@ -5617,16 +5725,10 @@ impl Sets {
             futures.push(async move { set.list_path(rx_clone, opts, send).await });
         }
 
-        tokio::spawn(
-            async move {
-                if let Err(err) = merge_entry_channels(rx, inputs, sender.clone(), 1).await {
-                    error!("merge_entry_channels err {:?}", err);
-                }
-            }
-            .instrument(tracing::Span::current()),
-        );
+        let merge_task = spawn_listing_merge(rx, inputs, sender);
 
         let results = join_all(futures).await;
+        merge_task.await.map_err(Error::from)??;
         let mut all_at_eof = true;
         let mut errs = Vec::new();
         for result in results {
@@ -5677,6 +5779,7 @@ impl Sets {
     ) -> Result<()> {
         check_list_objs_args(bucket, prefix, &None)?;
 
+        let rx = rx.child_token();
         let mut futures = Vec::new();
         let mut inputs = Vec::new();
 
@@ -6007,17 +6110,11 @@ impl Sets {
             .instrument(tracing::Span::current()),
         );
 
-        tokio::spawn(
-            async move {
-                if let Err(err) = merge_entry_channels(rx, inputs, merge_tx, 1).await {
-                    error!("merge_entry_channels err {:?}", err)
-                }
-            }
-            .instrument(tracing::Span::current()),
-        );
+        let merge_task = spawn_listing_merge(rx, inputs, merge_tx);
 
         let walk_started = std::time::Instant::now();
         let walk_results = join_all(futures).await;
+        merge_task.await.map_err(Error::from)??;
         let mut errs = Vec::new();
         for walk_result in walk_results {
             match walk_result {
@@ -7016,7 +7113,7 @@ mod test {
     };
     use crate::cache_value::metacache_set::{FallbackClaimTracker, TestReaderBehavior, list_path_raw};
     use crate::disk::{DiskAPI, DiskOption, STORAGE_FORMAT_FILE, endpoint::Endpoint, error::DiskError, new_disk};
-    use crate::error::StorageError;
+    use crate::error::{Result, StorageError};
     use crate::object_api::ObjectInfo;
     use rustfs_filemeta::{
         FileInfo, FileMeta, FileMetaVersion, MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntry, MetaDeleteMarker,
@@ -10399,7 +10496,7 @@ mod test {
     }
 
     #[tokio::test]
-    async fn merge_entry_channels_documents_candidate_metadata_authority_risk() {
+    async fn merge_entry_channels_preserves_cross_pool_delete_marker_versions() {
         let (tx_a, rx_a) = mpsc::channel(4);
         let (tx_b, rx_b) = mpsc::channel(4);
         let (tx_c, rx_c) = mpsc::channel(4);
@@ -10428,9 +10525,13 @@ mod test {
             .expect("merged entry should be present");
         assert_eq!(merged.name, "obj-a");
         assert!(
-            !merged.is_latest_delete_marker(),
-            "current merge consumes candidate metadata bytes; future index-backed strong modes must live-verify metadata instead"
+            merged.is_latest_delete_marker(),
+            "a newer marker must remain current across independently resolved pools"
         );
+        let versions = merged.file_info_versions("bucket").expect("merged versions should decode");
+        assert_eq!(versions.versions.len(), 2, "retain the historical object and deduplicate the marker");
+        assert!(versions.versions[0].deleted && versions.versions[0].is_latest);
+        assert!(!versions.versions[1].deleted && !versions.versions[1].is_latest);
         assert!(
             matches!(timeout(Duration::from_secs(1), out_rx.recv()).await, Ok(None)),
             "merge should not emit a duplicate entry for the same key"
@@ -10442,6 +10543,250 @@ mod test {
             .expect("merge task should succeed");
     }
 
+    fn rewrite_test_version(mut entry: MetaCacheEntry, change: impl FnOnce(&mut FileMetaVersion)) -> MetaCacheEntry {
+        let meta = entry.cached.as_mut().expect("test metadata should be decoded");
+        assert_eq!(meta.versions.len(), 1);
+        let mut version = meta.versions[0].parse_version_meta().expect("test version should decode");
+        change(&mut version);
+        meta.versions[0] = version.try_into().expect("test version should encode");
+        entry.metadata = meta.marshal_msg().expect("test metadata should encode");
+        entry
+    }
+
+    async fn merge_test_object_entries(entries: Vec<MetaCacheEntry>) -> Result<MetaCacheEntry> {
+        let mut inputs = Vec::with_capacity(entries.len());
+        for entry in entries {
+            let (sender, receiver) = mpsc::channel(1);
+            sender.send(entry).await.expect("fixture entry should queue");
+            inputs.push(receiver);
+        }
+        let (sender, mut receiver) = mpsc::channel(1);
+        let task = tokio::spawn(merge_entry_channels(CancellationToken::new(), inputs, sender, 1));
+        let entry = receiver.recv().await;
+        task.await.expect("merge must not panic")?;
+        Ok(entry.expect("a valid same-key group must produce an entry"))
+    }
+
+    #[tokio::test]
+    async fn merge_entry_channels_orders_complete_histories_independently_of_pool_order() {
+        let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let first = test_object_meta_entry_with_erasure_versions("key", &[(time, "first", 4, 2)]);
+        let second = rewrite_test_version(
+            test_object_meta_entry_with_erasure_versions("key", &[(time, "second", 4, 2)]),
+            |version| version.object.as_mut().expect("object version").version_id = Some(Uuid::from_u128(2)),
+        );
+        let marker = test_delete_marker_meta_entry("key", time + time::Duration::seconds(1));
+        let inputs = [first, second, marker];
+        let mut expected = None;
+        for order in [[0, 1, 2], [0, 2, 1], [1, 0, 2], [1, 2, 0], [2, 0, 1], [2, 1, 0]] {
+            let entry = merge_test_object_entries(order.map(|index| inputs[index].clone()).to_vec())
+                .await
+                .expect("disjoint version chains should merge");
+            let versions = entry.file_info_versions("bucket").expect("merged versions should decode");
+            assert_eq!(versions.versions.len(), 3);
+            assert!(versions.versions[0].deleted && versions.versions[0].is_latest);
+            assert!(
+                versions.versions[1..]
+                    .iter()
+                    .all(|version| !version.deleted && !version.is_latest)
+            );
+            assert!(versions.versions.iter().all(|version| version.num_versions == 3));
+            let identities = versions.versions.iter().map(|version| version.version_id).collect::<Vec<_>>();
+            assert!(identities.contains(&Some(Uuid::from_u128(1))));
+            assert!(identities.contains(&Some(Uuid::from_u128(2))));
+            if let Some(expected) = &expected {
+                assert_eq!(&identities, expected, "equal-time versions must have stable pagination order");
+            } else {
+                expected = Some(identities);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn merge_entry_channels_key_only_candidates_do_not_override_version_metadata() {
+        let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let marker = test_delete_marker_meta_entry("key", time);
+        for entries in [
+            vec![test_meta_entry("key"), marker.clone()],
+            vec![marker.clone(), test_meta_entry("key")],
+        ] {
+            let mut merged = merge_test_object_entries(entries)
+                .await
+                .expect("merge a name with resolved metadata");
+            assert!(merged.is_latest_delete_marker());
+            assert_eq!(merged.file_info_versions("bucket").expect("decode marker").versions.len(), 1);
+        }
+    }
+
+    #[tokio::test]
+    async fn merge_entry_channels_accepts_equivalent_migrated_coding_and_data_dirs() {
+        let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let first = test_object_meta_entry_with_erasure_versions("key", &[(time, "same-etag", 4, 2)]);
+        let second = rewrite_test_version(
+            test_object_meta_entry_with_erasure_versions("key", &[(time, "same-etag", 6, 2)]),
+            |version| version.object.as_mut().expect("object version").data_dir = Some(Uuid::from_u128(42)),
+        );
+        let forward = merge_test_object_entries(vec![first.clone(), second.clone()])
+            .await
+            .expect("valid migration copies");
+        let reverse = merge_test_object_entries(vec![second, first])
+            .await
+            .expect("reversed migration copies");
+        assert_eq!(forward.metadata, reverse.metadata, "representation must not depend on channel order");
+        let versions = forward.file_info_versions("bucket").expect("merged metadata should decode");
+        assert_eq!(versions.versions.len(), 1);
+        assert_eq!(versions.versions[0].metadata.get("etag").map(String::as_str), Some("same-etag"));
+    }
+
+    #[tokio::test]
+    async fn merge_entry_channels_rejects_conflicting_version_identity_and_metadata() {
+        let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let original = test_object_meta_entry_with_erasure_versions("key", &[(time, "original", 4, 2)]);
+        for key in [
+            "etag",
+            "x-amz-tagging",
+            "x-amz-object-lock-mode",
+            "x-amz-object-lock-retain-until-date",
+        ] {
+            let changed = rewrite_test_version(original.clone(), |version| {
+                version
+                    .object
+                    .as_mut()
+                    .expect("object version")
+                    .meta_user
+                    .insert(key.to_string(), "changed".to_string());
+            });
+            for pair in [[original.clone(), changed.clone()], [changed, original.clone()]] {
+                let err = merge_test_object_entries(pair.to_vec())
+                    .await
+                    .expect_err("conflicting copies must fail");
+                assert_eq!(err, StorageError::FileCorrupt, "conflict in {key} must not become arbitrary metadata");
+            }
+        }
+        let marker = rewrite_test_version(test_delete_marker_meta_entry("key", time), |version| {
+            version.delete_marker.as_mut().expect("delete marker").version_id = Some(Uuid::from_u128(1));
+        });
+        assert_eq!(
+            merge_test_object_entries(vec![original, marker])
+                .await
+                .expect_err("UUID type conflict"),
+            StorageError::FileCorrupt
+        );
+    }
+
+    #[tokio::test]
+    async fn merge_entry_channels_reconciles_null_overwrite_without_losing_uuid_history() {
+        let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let history = test_object_meta_entry_with_erasure_versions("key", &[(time, "history", 4, 2)]);
+        let old_null = rewrite_test_version(history.clone(), |version| {
+            version.object.as_mut().expect("null object").version_id = None;
+        });
+        let marker = rewrite_test_version(test_delete_marker_meta_entry("key", time + time::Duration::seconds(1)), |version| {
+            version.delete_marker.as_mut().expect("null marker").version_id = Some(Uuid::nil());
+        });
+        for inputs in [
+            vec![old_null.clone(), marker.clone(), history.clone()],
+            vec![history, marker, old_null],
+        ] {
+            let entry = merge_test_object_entries(inputs)
+                .await
+                .expect("new null slot should replace old null slot");
+            let versions = entry.file_info_versions("bucket").expect("null versions should decode");
+            assert_eq!(versions.versions.len(), 2);
+            assert!(versions.versions[0].deleted && versions.versions[0].is_latest);
+            assert!(versions.versions[0].version_id.is_none_or(|id| id.is_nil()));
+            assert_eq!(versions.versions[1].version_id, Some(Uuid::from_u128(1)));
+        }
+    }
+
+    #[tokio::test]
+    async fn merge_entry_channels_rejects_corrupt_version_headers_and_empty_stacks() {
+        let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let original = test_object_meta_entry_with_erasure_versions("key", &[(time, "etag", 4, 2)]);
+        for empty in [false, true] {
+            let mut corrupt = original.clone();
+            let meta = corrupt.cached.as_mut().expect("fixture metadata");
+            if empty {
+                meta.versions.clear();
+            } else {
+                meta.versions[0].header.version_id = Some(Uuid::from_u128(99));
+            }
+            corrupt.metadata = meta.marshal_msg().expect("encode corrupt fixture");
+            assert_eq!(
+                merge_test_object_entries(vec![original.clone(), corrupt])
+                    .await
+                    .expect_err("corrupt candidate must fail"),
+                StorageError::FileCorrupt
+            );
+        }
+        let mut malformed = original.clone();
+        malformed.cached = None;
+        malformed.metadata = vec![0xff];
+        assert_eq!(
+            merge_test_object_entries(vec![original, malformed])
+                .await
+                .expect_err("malformed metadata must fail"),
+            StorageError::FileCorrupt
+        );
+    }
+
+    #[tokio::test]
+    async fn merge_entry_channels_does_not_combine_subquorum_markers_across_erasure_sets() {
+        let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let old = test_object_meta_entry_with_erasure_versions("key", &[(time, "history", 4, 2)]);
+        let marked = test_object_with_delete_marker_meta_entry("key", time, time + time::Duration::seconds(1));
+        let mut inputs = Vec::new();
+        for marker_copies in [1, 2] {
+            let resolver = list_metadata_resolution_params("bucket".to_string(), 3, 3, true, 0);
+            let copies = (0..3)
+                .map(|index| Some(if index < marker_copies { marked.clone() } else { old.clone() }))
+                .collect();
+            let entry = resolve_listing_entries(MetaCacheEntries(copies), resolver, false)
+                .expect("each set independently retains its quorum-backed history");
+            inputs.push(entry);
+        }
+        let merged = merge_test_object_entries(inputs).await.expect("merge resolved histories");
+        let versions = merged.file_info_versions("bucket").expect("decode merged history");
+        assert_eq!(
+            versions.versions.len(),
+            1,
+            "three marker copies across two EC domains do not form a quorum"
+        );
+        assert!(!versions.versions[0].deleted);
+    }
+
+    #[tokio::test]
+    async fn listing_merge_preserves_error_after_partial_output_without_cancelling_request() {
+        let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let (first_tx, first_rx) = mpsc::channel(2);
+        let (second_tx, second_rx) = mpsc::channel(1);
+        first_tx.send(test_meta_entry("a/")).await.expect("queue preceding prefix");
+        first_tx
+            .send(test_object_meta_entry_with_erasure_versions("b", &[(time, "one", 4, 2)]))
+            .await
+            .expect("queue first copy");
+        second_tx
+            .send(test_object_meta_entry_with_erasure_versions("b", &[(time, "two", 4, 2)]))
+            .await
+            .expect("queue conflicting copy");
+        drop(first_tx);
+        drop(second_tx);
+        let request = CancellationToken::new();
+        let workers = request.child_token();
+        let (sender, mut receiver) = mpsc::channel(1);
+        let task = super::spawn_listing_merge(workers.clone(), vec![first_rx, second_rx], sender);
+        assert_eq!(receiver.recv().await.expect("preceding result should arrive").name, "a/");
+        assert!(receiver.recv().await.is_none());
+        assert_eq!(
+            task.await
+                .expect("merge task must not panic")
+                .expect_err("conflict must propagate"),
+            StorageError::FileCorrupt
+        );
+        assert!(workers.is_cancelled(), "failed merge must stop the disk producers");
+        assert!(!request.is_cancelled(), "the API must still observe the merge error");
+    }
+
     #[tokio::test]
     async fn merge_entry_channels_handles_single_channel() {
         let (tx, rx) = mpsc::channel(4);

crates/ecstore/src/store/object.rs

@@ -2013,7 +2013,7 @@ fn effective_object_actual_size(info: &ObjectInfo) -> Option<i64> {
     info.get_actual_size().ok()
 }
 
-fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &ObjectInfo) -> bool {
+pub(super) fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &ObjectInfo) -> bool {
     is_data_movement_delete_marker(source)
         && is_data_movement_delete_marker(target)
         && source.version_id == target.version_id
@@ -4791,7 +4791,13 @@ impl ECStore {
             return Ok(ObjectInfo::default());
         }
 
-        let gopts = delete_pool_lookup_opts(&opts, true);
+        let creates_latest_marker = should_create_delete_marker_for_missing_object(&opts);
+        let mut gopts = delete_pool_lookup_opts(&opts, true);
+        if creates_latest_marker {
+            // An unwritable source still owns its current version. Hiding it
+            // during lookup would turn a rejected write into a new-pool marker.
+            gopts.skip_rebalancing = false;
+        }
 
         if opts.data_movement {
             let existing_pool_info = self.get_pool_info_existing_with_opts(bucket, object, &gopts).await;
@@ -4917,7 +4923,12 @@ impl ECStore {
         }
 
         // Determine which pool contains it
-        let (mut pinfo, errs) = match self.get_pool_info_existing_with_opts(bucket, object, &gopts).await {
+        let existing_pool_info = if creates_latest_marker {
+            self.get_pool_info_for_delete_marker(bucket, object, &gopts).await
+        } else {
+            self.get_pool_info_existing_with_opts(bucket, object, &gopts).await
+        };
+        let (mut pinfo, errs) = match existing_pool_info {
             Ok(res) => res,
             Err(err) if is_err_read_quorum(&err) => return Err(StorageError::ErasureWriteQuorum),
             Err(err) if is_err_object_not_found(&err) && should_create_delete_marker_for_missing_object(&opts) => {
@@ -4954,7 +4965,7 @@ impl ECStore {
             }
         };
 
-        if pinfo.object_info.delete_marker && opts.version_id.is_none() {
+        if pinfo.object_info.delete_marker && opts.version_id.is_none() && !creates_latest_marker {
             pinfo.object_info.name = decode_dir_object(object);
             return Ok(pinfo.object_info);
         }
@@ -4976,7 +4987,13 @@ impl ECStore {
         }
 
         for pool in self.pools.iter() {
+            if creates_latest_marker && pool.pool_idx != pinfo.index {
+                continue;
+            }
             if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await {
+                if creates_latest_marker {
+                    return Err(StorageError::SlowDown);
+                }
                 continue;
             }
 
@@ -5001,7 +5018,7 @@ impl ECStore {
                     return Ok(obj);
                 }
                 Err(err) => {
-                    if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
+                    if creates_latest_marker || (!is_err_object_not_found(&err) && !is_err_version_not_found(&err)) {
                         return Err(err);
                     }
                 }
@@ -8291,6 +8308,546 @@ mod tests {
         }
     }
 
+    async fn multipool_version_test_store(bucket: &str) -> (Vec<tempfile::TempDir>, Arc<ECStore>) {
+        let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        let (mut dirs, first_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await;
+        let (second_dirs, second_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await;
+        dirs.extend(second_dirs);
+        let store = Arc::new(new_prepared_reader_test_store_with_ctx(&[first_set, second_set], ctx).await);
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+        store
+            .handle_make_bucket(
+                bucket,
+                &MakeBucketOptions {
+                    versioning_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create the versioned bucket in both pools");
+        (dirs, store)
+    }
+
+    #[tokio::test]
+    async fn multipool_delete_marker_stays_with_existing_versions() {
+        let bucket = "multipool-marker-routing";
+        let object = "history.bin";
+        let (_dirs, store) = multipool_version_test_store(bucket).await;
+
+        let mut expected_versions = Vec::new();
+        for value in 1..=3_u8 {
+            let written = store.pools[1]
+                .put_object(
+                    bucket,
+                    object,
+                    &mut PutObjReader::from_vec(vec![value; 4097]),
+                    &ObjectOptions {
+                        versioned: true,
+                        user_defined: HashMap::from([
+                            (rustfs_utils::http::AMZ_OBJECT_TAGGING.to_string(), format!("generation={value}")),
+                            ("x-amz-meta-generation".to_string(), value.to_string()),
+                        ]),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("write a historical version deterministically to pool 1");
+            expected_versions.push(written.version_id.expect("versioned PUT must acknowledge a UUID"));
+        }
+        let marker = store
+            .delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("delete the current version");
+        assert!(marker.delete_marker);
+        let marker_id = marker.version_id.expect("DELETE must acknowledge a marker UUID");
+        assert!(!expected_versions.contains(&marker_id));
+
+        let local = store.pools[1]
+            .clone()
+            .inner_list_object_versions(bucket, object, None, None, None, 10)
+            .await
+            .expect("list the object-owning pool");
+        assert_eq!(local.objects.len(), 4, "the marker must be committed beside the three existing versions");
+        assert_eq!(local.objects[0].version_id, Some(marker_id));
+        assert!(local.objects[0].delete_marker && local.objects[0].is_latest);
+        let versions = store
+            .clone()
+            .inner_list_object_versions(bucket, object, None, None, None, 10)
+            .await
+            .expect("list all pools");
+        assert_eq!(versions.objects.len(), 4);
+        assert_eq!(versions.objects.iter().filter(|version| version.is_latest).count(), 1);
+        for (index, version_id) in expected_versions.iter().copied().enumerate() {
+            assert!(
+                versions
+                    .objects
+                    .iter()
+                    .any(|version| version.version_id == Some(version_id) && !version.is_latest)
+            );
+            let mut reader = store
+                .handle_get_object_reader(
+                    bucket,
+                    object,
+                    None,
+                    HeaderMap::new(),
+                    &ObjectOptions {
+                        version_id: Some(version_id.to_string()),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("historical version should remain readable");
+            let mut payload = Vec::new();
+            reader
+                .stream
+                .read_to_end(&mut payload)
+                .await
+                .expect("read all historical bytes");
+            let value = u8::try_from(index + 1).expect("fixture generation fits u8");
+            assert_eq!(payload, vec![value; 4097]);
+            let listed = versions
+                .objects
+                .iter()
+                .find(|version| version.version_id == Some(version_id))
+                .expect("listed historical version");
+            assert_eq!(listed.user_tags.as_str(), format!("generation={value}"));
+            assert_eq!(listed.user_defined.get("x-amz-meta-generation"), Some(&value.to_string()));
+        }
+
+        let repeated = store
+            .delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("repeat simple DELETE");
+        assert!(repeated.delete_marker);
+        assert_ne!(
+            repeated.version_id,
+            Some(marker_id),
+            "each enabled-versioning DELETE creates a new marker"
+        );
+        store
+            .delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    version_id: repeated.version_id.map(|id| id.to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("remove the newest marker by identity");
+        store
+            .delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    version_id: Some(marker_id.to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("remove the original marker by identity");
+        let current = store
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("previous version becomes current");
+        assert_eq!(current.version_id, expected_versions.last().copied());
+        assert!(!current.delete_marker);
+    }
+
+    #[tokio::test]
+    async fn multipool_existing_split_history_lists_and_paginates_without_metadata_writes() {
+        for marker_pool in 0..2 {
+            let bucket = format!("multipool-split-history-{marker_pool}");
+            let object = "history.bin";
+            let (dirs, store) = multipool_version_test_store(&bucket).await;
+            let mut expected = Vec::new();
+            for value in 1..=3_u8 {
+                let version = store.pools[1 - marker_pool]
+                    .put_object(
+                        &bucket,
+                        object,
+                        &mut PutObjReader::from_vec(vec![value; 4097]),
+                        &ObjectOptions {
+                            versioned: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("seed history through the owning pool's normal write path");
+                expected.push(version.version_id);
+            }
+            let marker = store.pools[marker_pool]
+                .delete_object(
+                    &bucket,
+                    object,
+                    ObjectOptions {
+                        versioned: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("reproduce the previously committed split marker with a normal pool DELETE");
+            assert!(marker.delete_marker);
+            expected.push(marker.version_id);
+            expected.reverse();
+            let mut before = Vec::new();
+            for dir in &dirs {
+                before.push(
+                    tokio::fs::read(dir.path().join(&bucket).join(object).join("xl.meta"))
+                        .await
+                        .expect("snapshot persisted version metadata"),
+                );
+            }
+            for max_keys in [1, 2, 4, 10] {
+                let mut key_marker = None;
+                let mut version_marker = None;
+                let mut listed = Vec::new();
+                let mut completed = false;
+                for _ in 0..6 {
+                    let page = store
+                        .clone()
+                        .inner_list_object_versions(&bucket, object, key_marker.clone(), version_marker.clone(), None, max_keys)
+                        .await
+                        .expect("read a complete merged version page");
+                    listed.extend(page.objects);
+                    if !page.is_truncated {
+                        completed = true;
+                        break;
+                    }
+                    assert_ne!(
+                        (&page.next_marker, &page.next_version_idmarker),
+                        (&key_marker, &version_marker),
+                        "version cursor must advance"
+                    );
+                    key_marker = page.next_marker;
+                    version_marker = page.next_version_idmarker;
+                }
+                assert!(completed, "pagination must terminate");
+                assert_eq!(listed.iter().map(|version| version.version_id).collect::<Vec<_>>(), expected);
+                assert!(listed[0].delete_marker && listed[0].is_latest);
+                assert!(listed[1..].iter().all(|version| !version.delete_marker && !version.is_latest));
+            }
+            let visible = store
+                .clone()
+                .list_objects_generic(&bucket, "", None, None, 10, false)
+                .await
+                .expect("list current objects");
+            assert!(visible.objects.is_empty(), "the global current marker hides the object");
+            for (dir, before) in dirs.iter().zip(before) {
+                assert_eq!(
+                    tokio::fs::read(dir.path().join(&bucket).join(object).join("xl.meta"))
+                        .await
+                        .expect("read unchanged metadata"),
+                    before
+                );
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn multipool_conflicting_version_metadata_fails_the_listing_request() {
+        let bucket = "multipool-version-conflict";
+        let (_dirs, store) = multipool_version_test_store(bucket).await;
+        store.pools[0]
+            .put_object(
+                bucket,
+                "a.bin",
+                &mut PutObjReader::from_vec(b"preceding result".to_vec()),
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed an entry before the conflict");
+        let version_id = Uuid::new_v4();
+        let mod_time = OffsetDateTime::now_utc();
+        let mut etags = Vec::new();
+        for (pool_idx, value) in [(0, 1), (1, 2)] {
+            let written = store.pools[pool_idx]
+                .put_object(
+                    bucket,
+                    "z.bin",
+                    &mut PutObjReader::from_vec(vec![value; 4097]),
+                    &ObjectOptions {
+                        versioned: true,
+                        version_id: Some(version_id.to_string()),
+                        mod_time: Some(mod_time),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("persist independent conflicting copies");
+            assert_eq!(written.version_id, Some(version_id));
+            etags.push(written.etag);
+        }
+        assert_ne!(etags[0], etags[1], "fixture must contain a semantic conflict");
+        let err = store
+            .clone()
+            .inner_list_object_versions(bucket, "", None, None, None, 10)
+            .await
+            .expect_err("partial output must not hide the merge error");
+        assert_eq!(err, StorageError::FileCorrupt);
+
+        let cancellation = tokio_util::sync::CancellationToken::new();
+        let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
+        let walk = store
+            .clone()
+            .walk(cancellation.clone(), bucket, "", sender, WalkOptions::default());
+        let drain = async { while receiver.recv().await.is_some() {} };
+        let (walk_result, ()) = tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(walk, drain) })
+            .await
+            .expect("bounded walk output must drain and terminate on a merge error");
+        assert_eq!(
+            walk_result.expect_err("walk must report the same metadata conflict"),
+            StorageError::FileCorrupt
+        );
+        assert!(!cancellation.is_cancelled(), "worker failure must not cancel the caller's request");
+    }
+
+    #[tokio::test]
+    async fn multipool_marker_rejects_unwritable_owner_without_falling_back() {
+        let bucket = "multipool-unwritable-owner";
+        let object = "history.bin";
+        let (_dirs, store) = multipool_version_test_store(bucket).await;
+        store.pools[1]
+            .put_object(
+                bucket,
+                object,
+                &mut PutObjReader::from_vec(vec![1; 4097]),
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed the nonzero owner");
+        *store.pool_meta.write().await = PoolMeta {
+            pools: vec![prepared_pool_test_status(0, false), prepared_pool_test_status(1, true)],
+            ..Default::default()
+        };
+        let err = store
+            .delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("suspended owner must reject marker creation");
+        assert_eq!(err, StorageError::SlowDown);
+        *store.pool_meta.write().await = PoolMeta::default();
+
+        let mut rebalancing = crate::services::rebalance::RebalanceStats {
+            participating: true,
+            ..Default::default()
+        };
+        rebalancing.info.status = crate::services::rebalance::RebalStatus::Started;
+        *store.rebalance_meta.write().await = Some(crate::services::rebalance::RebalanceMeta {
+            pool_stats: vec![crate::services::rebalance::RebalanceStats::default(), rebalancing],
+            ..Default::default()
+        });
+        let error = store
+            .delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await;
+        *store.rebalance_meta.write().await = None;
+        assert_eq!(
+            error.expect_err("rebalance must not hide the owner during marker lookup"),
+            StorageError::SlowDown
+        );
+
+        // Isolate object quorum from the bucket metadata preflight: disabling
+        // a whole pool can otherwise fail before object ownership is looked up.
+        let quorum_bucket = RUSTFS_META_BUCKET;
+        store.pools[1]
+            .put_object(
+                quorum_bucket,
+                object,
+                &mut PutObjReader::from_vec(vec![1; 4097]),
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed the object-quorum fixture");
+        for pool in &store.pools {
+            pool.put_object(
+                quorum_bucket,
+                "split.bin",
+                &mut PutObjReader::from_vec(vec![2; 4097]),
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed another key with history in both pools");
+        }
+
+        let owner = &store.pools[1].disk_set[0];
+        let healthy = owner.disks.read().await.clone();
+        for disk in owner.disks.write().await.iter_mut().skip(1) {
+            *disk = None;
+        }
+        let error = store
+            .delete_object(
+                quorum_bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await;
+        let split_error = store
+            .delete_object(
+                quorum_bucket,
+                "split.bin",
+                ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await;
+        *owner.disks.write().await = healthy;
+        assert_eq!(
+            error.expect_err("subquorum owner must not become a new-pool marker"),
+            StorageError::ErasureWriteQuorum
+        );
+        assert_eq!(
+            split_error.expect_err("a readable older pool does not prove the global current version"),
+            StorageError::ErasureWriteQuorum
+        );
+        let split = store.pools[0]
+            .clone()
+            .inner_list_object_versions(quorum_bucket, "split.bin", None, None, None, 10)
+            .await
+            .expect("inspect the readable older pool");
+        assert_eq!(split.objects.len(), 1);
+        assert!(!split.objects[0].delete_marker);
+        let other = store.pools[0]
+            .clone()
+            .inner_list_object_versions(quorum_bucket, object, None, None, None, 10)
+            .await
+            .expect("inspect the other pool");
+        assert!(other.objects.is_empty());
+        let history = store.pools[1]
+            .clone()
+            .inner_list_object_versions(quorum_bucket, object, None, None, None, 10)
+            .await
+            .expect("inspect the restored owner");
+        assert_eq!(history.objects.len(), 1);
+        assert!(!history.objects[0].delete_marker);
+    }
+
+    #[tokio::test]
+    async fn multipool_suspended_and_batch_markers_keep_the_null_slot_on_the_owner() {
+        let bucket = "multipool-suspended-markers";
+        let object = "history.bin";
+        let (_dirs, store) = multipool_version_test_store(bucket).await;
+        let original = store.pools[1]
+            .put_object(
+                bucket,
+                object,
+                &mut PutObjReader::from_vec(vec![1; 4097]),
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("seed the UUID history in pool 1");
+        store
+            .update_bucket_metadata_config(
+                bucket,
+                crate::bucket::metadata::BUCKET_VERSIONING_CONFIG,
+                b"<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>".to_vec(),
+            )
+            .await
+            .expect("persist suspended bucket versioning");
+        let null = store
+            .put_object(
+                bucket,
+                object,
+                &mut PutObjReader::from_vec(vec![2; 4097]),
+                &ObjectOptions {
+                    version_suspended: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("write the null version beside the history");
+        assert!(null.version_id.is_none_or(|id| id.is_nil()));
+        for _ in 0..2 {
+            let marker = store
+                .delete_object(
+                    bucket,
+                    object,
+                    ObjectOptions {
+                        version_suspended: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("replace the null slot with a marker");
+            assert!(marker.delete_marker);
+            assert!(marker.version_id.is_none_or(|id| id.is_nil()));
+        }
+        let (deleted, errors) = store
+            .delete_objects(
+                bucket,
+                vec![ObjectToDelete {
+                    object_name: object.to_string(),
+                    ..Default::default()
+                }],
+                ObjectOptions::default(),
+            )
+            .await;
+        assert!(errors.iter().all(Option::is_none), "batch DELETE should succeed: {errors:?}");
+        assert_eq!(deleted.len(), 1);
+        assert!(deleted[0].delete_marker);
+        let versions = store.pools[1]
+            .clone()
+            .inner_list_object_versions(bucket, object, None, None, None, 10)
+            .await
+            .expect("inspect the owner after batch DELETE");
+        assert_eq!(versions.objects.len(), 2, "only one null marker and the UUID history remain");
+        assert!(versions.objects[0].delete_marker && versions.objects[0].is_latest);
+        assert_eq!(versions.objects[1].version_id, original.version_id);
+        let other = store.pools[0]
+            .clone()
+            .inner_list_object_versions(bucket, object, None, None, None, 10)
+            .await
+            .expect("inspect the unused pool");
+        assert!(other.objects.is_empty());
+    }
+
     async fn assert_prepared_reader_blocks_writer(store: &ECStore, bucket: &str, object: &str) {
         assert_pool_writer_is_blocked(store, 0, bucket, object).await;
     }

crates/ecstore/src/store/rebalance.rs

@@ -611,7 +611,18 @@ impl ECStore {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<(PoolObjInfo, Vec<PoolErr>)> {
-        self.internal_get_pool_info_existing_with_opts(bucket, object, opts).await
+        self.internal_get_pool_info_existing_with_opts(bucket, object, opts, false)
+            .await
+    }
+
+    pub(super) async fn get_pool_info_for_delete_marker(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+    ) -> Result<(PoolObjInfo, Vec<PoolErr>)> {
+        self.internal_get_pool_info_existing_with_opts(bucket, object, opts, true)
+            .await
     }
 
     async fn internal_get_pool_info_existing_with_opts(
@@ -619,6 +630,7 @@ impl ECStore {
         bucket: &str,
         object: &str,
         opts: &ObjectOptions,
+        require_all_pool_reads: bool,
     ) -> Result<(PoolObjInfo, Vec<PoolErr>)> {
         let mut futures = Vec::new();
         for pool in self.pools.iter() {
@@ -647,6 +659,15 @@ impl ECStore {
                     });
                 }
                 Err(e) => {
+                    // A readable older pool cannot prove ownership of the
+                    // current version while another pool is unreadable. Check
+                    // both raw and object-scoped quorum errors before sorting.
+                    if require_all_pool_reads && !is_err_object_not_found(&e) && !is_err_version_not_found(&e) {
+                        return Err(match e {
+                            Error::ErasureReadQuorum | Error::InsufficientReadQuorum(_, _) => Error::ErasureWriteQuorum,
+                            err => err,
+                        });
+                    }
                     ress.push(PoolObjInfo {
                         index,
                         err: Some(e),