fix(ecstore): integrate verified rename preflight evidence

.config/ecstore-required-tests.json

@@ -0,0 +1,72 @@
+{
+  "lane": "ci/test-and-lint",
+  "tests": [
+    {
+      "invariant": "write-quorum",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::inline_put_commit_path_tests::inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one"
+    },
+    {
+      "invariant": "metadata-rollback",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::write_unique_file_info_reverts_metadata_when_write_quorum_fails"
+    },
+    {
+      "invariant": "stale-writer",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::put_object_tmp_cleanup_tests::put_object_no_lock_aborts_after_outer_namespace_lock_loss"
+    },
+    {
+      "invariant": "range-body",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::transition_upload_integrity_tests::transitioned_compressed_object_range_get_returns_plaintext_slice"
+    },
+    {
+      "invariant": "multipart-cancellation",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::multipart::tests::cancelled_complete_keeps_upload_lock_through_tail_cleanup"
+    },
+    {
+      "invariant": "list-uncommitted-version",
+      "suite": "rustfs-filemeta",
+      "name": "metacache::tests::resolve_with_write_quorum_slack_keeps_partial_latest_hidden_during_merge"
+    },
+    {
+      "invariant": "minio-object-fixture",
+      "suite": "rustfs-filemeta",
+      "name": "filemeta::test::parses_real_minio_object_xlmeta"
+    },
+    {
+      "invariant": "corrupt-part-arrays",
+      "suite": "rustfs-filemeta",
+      "name": "filemeta::test::crc_valid_but_part_arrays_corrupt_into_fileinfo_errors_not_panics"
+    }
+  ],
+  "fixtures": [
+    {
+      "path": "crates/filemeta/tests/fixtures/minio/object_large_bin.xlmeta.hex",
+      "sha256": "e8093767806d701e639b48d023190e858fbc4cde69bcfd83c22af8cba8452ce5",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/filemeta/tests/fixtures/minio/object_small_txt.xlmeta.hex",
+      "sha256": "2a415ad3a3be5a9440035d4026ff880e0e8c1ec1701be9f4e077734e8dce03da",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/filemeta/tests/fixtures/minio/object_versioned_txt.xlmeta.hex",
+      "sha256": "7f21f50c326dd8b0228deb6dbdb7052b3d0a3f8ee6c85d43486f0e6bb7a97261",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/ecstore/tests/fixtures/minio/bucket_metadata.blob.hex",
+      "sha256": "f2b6e260aff106adf6039feb1c645686e84e75404ff725491fb18668be5db203",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/ecstore/tests/fixtures/minio/bucket_metadata_full.xlmeta.hex",
+      "sha256": "3b6de589519c08a1614c8bd409bb8199c17d42043861b07bce513075e6fbfc12",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    }
+  ]
+}

.github/workflows/ci.yml

@@ -269,6 +269,7 @@ jobs:
           CARGO_BUILD_JOBS: ${{ (github.event_name == 'push' || github.event_name == 'workflow_dispatch') && '3' || '2' }}
         run: |
           mkdir -p artifacts/test-and-lint
+          rm -f target/nextest/ci/junit.xml
           ./scripts/ci/resource_sampler.sh start nextest
           trap './scripts/ci/resource_sampler.sh stop' EXIT
           set +e
@@ -277,6 +278,12 @@ jobs:
             --status-level all --final-status-level all \
             2>&1 | tee artifacts/test-and-lint/nextest.log
           status=${PIPESTATUS[0]}
+          if [[ "${status}" -eq 0 ]]; then
+            cargo nextest list --profile ci --all --exclude e2e_test --message-format json \
+              > artifacts/test-and-lint/core-test-listing.json \
+              && python3 scripts/check_test_wiring.py --check-core artifacts/test-and-lint/core-test-listing.json \
+              && test -s target/nextest/ci/junit.xml || status=$?
+          fi
           {
             echo "command=cargo nextest run --profile ci --all --exclude e2e_test"
             echo "exit_status=${status}"

crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs

@@ -1170,6 +1170,7 @@ pub async fn save_manual_transition_job_record_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()
@@ -1242,6 +1243,7 @@ pub(crate) async fn save_manual_transition_worker_result_if_absent(
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1270,6 +1272,7 @@ pub(crate) async fn save_manual_transition_task_if_absent(
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1621,6 +1624,7 @@ pub async fn save_manual_transition_scope_admission_if_absent(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1672,6 +1676,7 @@ pub async fn save_manual_transition_scope_admission_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -1733,6 +1733,7 @@ async fn save_config_if_none_fenced(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1832,6 +1833,7 @@ async fn save_decommission_manifest_checkpoint_if_match(
 
         let mut opts = ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             no_lock: true,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(observed_etag),
@@ -1960,6 +1962,7 @@ async fn save_config_if_match_fenced(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(etag.to_string()),
                 ..Default::default()
@@ -3780,6 +3783,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()
@@ -3869,6 +3873,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_match: Some(etag),
                         ..Default::default()
@@ -3893,6 +3898,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()

crates/ecstore/src/bucket/lifecycle/transition_transaction.rs

@@ -612,6 +612,7 @@ pub(crate) async fn save_transition_transaction_record(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -658,6 +659,7 @@ pub(crate) async fn save_transition_transaction_record_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(etag),
                 ..Default::default()

crates/ecstore/src/bucket/on_demand_migration/backfill.rs

@@ -684,6 +684,7 @@ async fn write_checkpoint(
     };
     let opts = ObjectOptions {
         max_parity: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         http_preconditions: Some(preconditions),
         ..Default::default()
     };

crates/ecstore/src/bucket/on_demand_migration/breaker.rs

@@ -86,7 +86,12 @@ impl BreakerVerdict {
             Some(SourceError::Throttled | SourceError::Timeout | SourceError::Connect(_) | SourceError::ServerError(_)) => {
                 BreakerVerdict::Failure
             }
-            Some(SourceError::AccessDenied | SourceError::Unsupported(_) | SourceError::Other(_)) => BreakerVerdict::Neutral,
+            Some(
+                SourceError::AccessDenied
+                | SourceError::Unsupported(_)
+                | SourceError::InvalidPagination(_)
+                | SourceError::Other(_),
+            ) => BreakerVerdict::Neutral,
         }
     }
 }

crates/ecstore/src/bucket/on_demand_migration/list_through.rs

@@ -189,8 +189,8 @@ pub enum SourceListPlan {
     /// delimiter — the source's own roll-up boundary matches the request's.
     Page { prefix: String },
     /// `filter.prefix` reaches past a delimiter, so every key the source could
-    /// contribute rolls into this one common prefix. One bounded probe listing
-    /// decides whether it exists; there is nothing to paginate.
+    /// contribute rolls into this one common prefix. Bounded probes follow
+    /// empty progressing pages until a key proves existence or the source ends.
     Folded { probe_prefix: String, common_prefix: String },
 }
 
@@ -279,6 +279,29 @@ pub struct FetchRequest {
     pub token: Option<String>,
 }
 
+/// Invalid pagination metadata. Opaque cursor values are never included in errors.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
+pub enum ListPageError {
+    #[error("truncated listing has no continuation token")]
+    Missing,
+    #[error("truncated listing has an empty continuation token")]
+    Empty,
+    #[error("truncated listing repeats a continuation token")]
+    Repeated,
+}
+
+pub(crate) fn validate_list_page(is_truncated: bool, token: Option<&str>, next_token: Option<&str>) -> Result<(), ListPageError> {
+    if is_truncated {
+        match next_token {
+            None => return Err(ListPageError::Missing),
+            Some("") => return Err(ListPageError::Empty),
+            Some(next) if Some(next) == token => return Err(ListPageError::Repeated),
+            Some(_) => {}
+        }
+    }
+    Ok(())
+}
+
 #[derive(Debug, Default)]
 struct SideState {
     start: SideCursor,
@@ -364,6 +387,11 @@ impl ListThroughMerger {
     /// or `filter.prefix` excludes it.
     pub fn disable_source(&mut self) {
         self.source.disabled = true;
+        // A refill can fail after a valid first page. A local-only response
+        // must discard both that source payload and its ordering horizon.
+        self.source.entries.clear();
+        self.source.pages.clear();
+        self.source.more = false;
     }
 
     pub fn next_fetch(&self) -> Option<FetchRequest> {
@@ -378,7 +406,13 @@ impl ListThroughMerger {
     /// Records one fetched page. `entries` must be sorted by `name` and already
     /// filtered with [`Self::accepts`]; the caller keeps the matching payloads
     /// in the same order.
-    pub fn push_page(&mut self, side: MergeSide, entries: Vec<ListEntryKey>, is_truncated: bool, next_token: Option<String>) {
+    pub fn push_page(
+        &mut self,
+        side: MergeSide,
+        entries: Vec<ListEntryKey>,
+        is_truncated: bool,
+        next_token: Option<String>,
+    ) -> Result<(), ListPageError> {
         let state = match side {
             MergeSide::Local => &mut self.local,
             MergeSide::Source => &mut self.source,
@@ -387,15 +421,19 @@ impl ListThroughMerger {
             Some(last) => last.next_token.clone(),
             None => state.start.token.clone(),
         };
-        // A truncated page without a cursor cannot be continued; treating the
-        // side as finished is the only alternative to looping on it forever.
-        state.more = is_truncated && next_token.is_some();
+        validate_list_page(is_truncated, token.as_deref(), next_token.as_deref())?;
+        // Also reject a cycle through an earlier page in this bounded fetch.
+        if is_truncated && state.pages.iter().any(|page| page.token == next_token) {
+            return Err(ListPageError::Repeated);
+        }
+        state.more = is_truncated;
         state.pages.push(FetchedPage {
             token,
             count: entries.len(),
             next_token: is_truncated.then_some(next_token).flatten(),
         });
         state.entries.extend(entries);
+        Ok(())
     }
 
     pub fn finish(self) -> MergeOutcome {
@@ -599,9 +637,15 @@ mod tests {
                 let (entries, truncated, next) = reference_page(keys, prefix, delimiter, fetch.token.as_deref(), max_keys);
                 let kept: Vec<ListEntryKey> = entries.into_iter().filter(|entry| merger.accepts(&entry.name)).collect();
                 buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.iter().cloned());
-                merger.push_page(fetch.side, kept, truncated, next);
+                merger
+                    .push_page(fetch.side, kept, truncated, next)
+                    .expect("reference provider pages must advance");
             }
             let outcome = merger.finish();
+            assert_eq!(outcome.is_truncated, outcome.next_token.is_some());
+            if outcome.is_truncated {
+                assert_ne!(outcome.next_token, token, "every truncated merged page must make progress");
+            }
             page_sizes.push(outcome.picks.len());
             for pick in &outcome.picks {
                 let entry = buffers[usize::from(pick.side == MergeSide::Source)][pick.index].clone();
@@ -616,11 +660,25 @@ mod tests {
     }
 
     fn expected(local: &[String], source: &[String], prefix: &str, delimiter: Option<&str>) -> Vec<ListEntryKey> {
-        let mut all: Vec<String> = local.iter().chain(source.iter()).cloned().collect();
-        all.sort();
-        all.dedup();
-        let (entries, _, _) = reference_page(&all, prefix, delimiter, None, usize::MAX);
-        entries
+        // This oracle builds the complete namespace independently of the
+        // provider's page/marker helper and the production merger.
+        let mut namespace = std::collections::BTreeMap::new();
+        for key in local.iter().chain(source) {
+            let Some(suffix) = key.strip_prefix(prefix) else {
+                continue;
+            };
+            if let Some(delimiter) = delimiter.filter(|delimiter| !delimiter.is_empty())
+                && let Some((directory, _)) = suffix.split_once(delimiter)
+            {
+                namespace.insert(format!("{prefix}{directory}{delimiter}"), true);
+                continue;
+            }
+            namespace.insert(key.clone(), false);
+        }
+        namespace
+            .into_iter()
+            .map(|(name, is_prefix)| ListEntryKey { name, is_prefix })
+            .collect()
     }
 
     #[test]
@@ -662,7 +720,9 @@ mod tests {
                 token: None
             })
         );
-        merger.push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None);
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None)
+            .expect("local EOF is valid");
         assert_eq!(merger.next_fetch(), None);
         let outcome = merger.finish();
         assert_eq!(outcome.picks.len(), 1);
@@ -683,12 +743,14 @@ mod tests {
         };
         let mut merger = ListThroughMerger::new(1, Some(&resume));
         merger.disable_source();
-        merger.push_page(
-            MergeSide::Local,
-            vec![ListEntryKey::object("b"), ListEntryKey::object("c")],
-            true,
-            Some("local-2".to_string()),
-        );
+        merger
+            .push_page(
+                MergeSide::Local,
+                vec![ListEntryKey::object("b"), ListEntryKey::object("c")],
+                true,
+                Some("local-2".to_string()),
+            )
+            .expect("local cursor advances");
         let outcome = merger.finish();
         assert!(outcome.is_truncated);
         let token = outcome.next_token.expect("truncated page carries a token");
@@ -698,6 +760,212 @@ mod tests {
         assert_eq!(token.local.as_deref(), Some("local-1"), "a partly read page is re-listed");
     }
 
+    #[test]
+    fn truncated_pages_require_a_nonempty_advancing_cursor() {
+        for side in [MergeSide::Local, MergeSide::Source] {
+            for entries in [vec![], vec![ListEntryKey::object("a")]] {
+                for (next, expected) in [
+                    (None, Err(ListPageError::Missing)),
+                    (Some(""), Err(ListPageError::Empty)),
+                    (Some("stuck"), Err(ListPageError::Repeated)),
+                    (Some("advances"), Ok(())),
+                ] {
+                    let resume = ListThroughToken::new(
+                        SideCursor {
+                            token: Some("stuck".into()),
+                            done: false,
+                        },
+                        SideCursor {
+                            token: Some("stuck".into()),
+                            done: false,
+                        },
+                        None,
+                    );
+                    let mut merger = ListThroughMerger::new(2, Some(&resume));
+                    let result = merger.push_page(side, entries.clone(), true, next.map(str::to_string));
+                    assert_eq!(result, expected, "{side:?}, {entries:?}, {next:?}");
+                    let state = if side == MergeSide::Local {
+                        &merger.local
+                    } else {
+                        &merger.source
+                    };
+                    assert_eq!(state.pages.len(), usize::from(result.is_ok()), "invalid page must not be accepted");
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn repeated_empty_cursor_is_rejected_before_an_identical_page_can_escape() {
+        let resume = ListThroughToken::new(
+            SideCursor { token: None, done: true },
+            SideCursor {
+                token: Some("stuck".into()),
+                done: false,
+            },
+            None,
+        );
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        assert_eq!(
+            merger.next_fetch(),
+            Some(FetchRequest {
+                side: MergeSide::Source,
+                token: Some("stuck".into())
+            })
+        );
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("stuck".into())),
+            Err(ListPageError::Repeated)
+        );
+    }
+
+    #[test]
+    fn empty_pages_may_advance_within_the_fetch_budget_until_eof() {
+        let mut merger = ListThroughMerger::new(2, None);
+        merger.push_page(MergeSide::Local, vec![], false, None).expect("local EOF");
+        for next in ["opaque-z", "opaque-a"] {
+            assert_eq!(merger.next_fetch().expect("bounded source fetch").side, MergeSide::Source);
+            merger
+                .push_page(MergeSide::Source, vec![], true, Some(next.into()))
+                .expect("opaque cursor advances regardless of sort order");
+        }
+        assert!(merger.next_fetch().is_none(), "two source fetches exhaust the request budget");
+        let outcome = merger.finish();
+        assert!(outcome.picks.is_empty());
+        assert!(outcome.is_truncated);
+        let token = outcome.next_token.expect("empty progressing page has a cursor");
+        assert_eq!(token.source.as_deref(), Some("opaque-a"));
+        let mut merger = ListThroughMerger::new(2, Some(&token));
+        assert_eq!(merger.next_fetch().expect("source resumes").token.as_deref(), Some("opaque-a"));
+        merger
+            .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], false, None)
+            .expect("source EOF");
+        let outcome = merger.finish();
+        assert_eq!(
+            outcome.picks,
+            vec![MergePick {
+                side: MergeSide::Source,
+                index: 0
+            }]
+        );
+        assert!(!outcome.is_truncated);
+        assert!(outcome.next_token.is_none());
+    }
+
+    #[test]
+    fn a_cursor_cycle_inside_the_fetch_budget_is_rejected() {
+        let resume = ListThroughToken::new(
+            SideCursor { token: None, done: true },
+            SideCursor {
+                token: Some("first".into()),
+                done: false,
+            },
+            None,
+        );
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        merger
+            .push_page(MergeSide::Source, vec![], true, Some("second".into()))
+            .expect("first page advances");
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("first".into())),
+            Err(ListPageError::Repeated)
+        );
+    }
+
+    #[test]
+    fn source_refill_failure_discards_buffered_source_entries_and_horizon() {
+        let mut merger = ListThroughMerger::new(2, None);
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::object("z")], false, None)
+            .expect("local EOF");
+        merger
+            .push_page(MergeSide::Source, vec![ListEntryKey::object("a")], true, Some("stuck".into()))
+            .expect("first source page advances");
+        assert_eq!(merger.next_fetch().expect("source refill is required").token.as_deref(), Some("stuck"));
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("stuck".into())),
+            Err(ListPageError::Repeated)
+        );
+        merger.disable_source();
+        let outcome = merger.finish();
+        assert_eq!(
+            outcome.picks,
+            vec![MergePick {
+                side: MergeSide::Local,
+                index: 0
+            }]
+        );
+        assert!(!outcome.is_truncated);
+        assert!(outcome.next_token.is_none());
+    }
+
+    #[test]
+    fn list_through_static_namespace_boundary_matrix() {
+        let corpus = [
+            "a",
+            "a/",
+            "a/b",
+            "a/b/child",
+            "a0",
+            "b",
+            "b/leaf",
+            "quote\"&<",
+            "space key",
+            "z",
+            "é",
+            "中/文",
+        ];
+        for count in [0, 1, 3, 4, corpus.len()] {
+            let keys: Vec<String> = corpus[..count].iter().map(|key| (*key).to_string()).collect();
+            for placement in 0..3 {
+                let (local, source): (Vec<_>, Vec<_>) =
+                    keys.iter()
+                        .enumerate()
+                        .fold((vec![], vec![]), |(mut local, mut source), (index, key)| {
+                            if placement != 1 || index % 2 == 0 {
+                                local.push(key.clone());
+                            }
+                            if placement != 0 || index % 2 == 0 {
+                                source.push(key.clone());
+                            }
+                            (local, source)
+                        });
+                for prefix in ["", "a", "a/", "中/"] {
+                    for delimiter in [None, Some("/")] {
+                        for max_keys in [1, 3, 4] {
+                            let oracle = expected(&local, &source, prefix, delimiter);
+                            let (emitted, sizes) = walk(&local, &source, prefix, delimiter, max_keys);
+                            assert_eq!(
+                                emitted.iter().map(|(entry, _)| entry.clone()).collect::<Vec<_>>(),
+                                oracle,
+                                "count={count}, placement={placement}, prefix={prefix}, delimiter={delimiter:?}, max={max_keys}"
+                            );
+                            let expected_sizes: Vec<_> = if oracle.is_empty() {
+                                vec![0]
+                            } else {
+                                oracle.chunks(max_keys).map(<[ListEntryKey]>::len).collect()
+                            };
+                            assert_eq!(sizes, expected_sizes, "exact max and max+1 boundaries must agree");
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn list_through_large_overlap_walk_keeps_all_5300_keys() {
+        let source: Vec<_> = (0..5000).map(|index| format!("k{index:05}")).collect();
+        let local: Vec<_> = (4800..5300).map(|index| format!("k{index:05}")).collect();
+        let (emitted, sizes) = walk(&local, &source, "", None, 333);
+        assert_eq!(emitted.len(), 5300);
+        for (index, (entry, side)) in emitted.iter().enumerate() {
+            assert_eq!(entry.name, format!("k{index:05}"));
+            assert_eq!(*side, if index >= 4800 { MergeSide::Local } else { MergeSide::Source });
+        }
+        assert_eq!(sizes, [vec![333; 15], vec![305]].concat());
+    }
+
     #[test]
     fn token_round_trips_and_rejects_tampering() {
         let token = ListThroughToken::new(
@@ -796,7 +1064,10 @@ mod tests {
     }
 
     proptest! {
-        #![proptest_config(ProptestConfig::with_cases(256))]
+        #![proptest_config(ProptestConfig {
+            rng_seed: proptest::test_runner::RngSeed::Fixed(0xec5706),
+            ..ProptestConfig::with_cases(256)
+        })]
 
         /// Full pagination of a merged listing equals the sorted, deduplicated
         /// union of both sides, with every shared key served by local, and no

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -25,6 +25,7 @@
 //! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never
 //! forwarded: v1 rejects SSE-C source objects outright.
 
+use super::list_through::{ListPageError, validate_list_page};
 use crate::bucket::remote_s3_client::{
     PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config,
 };
@@ -223,6 +224,8 @@ pub enum SourceError {
     ServerError(u16),
     #[error("unsupported source object: {0}")]
     Unsupported(String),
+    #[error("invalid source listing: {0}")]
+    InvalidPagination(#[from] ListPageError),
     #[error("source request failed: {0}")]
     Other(String),
 }
@@ -245,6 +248,7 @@ impl SourceError {
             SourceError::Connect(_) => "connect",
             SourceError::ServerError(_) => "server_error",
             SourceError::Unsupported(_) => "unsupported",
+            SourceError::InvalidPagination(_) => "invalid_pagination",
             SourceError::Other(_) => "other",
         }
     }
@@ -714,6 +718,7 @@ impl SourceClient {
                 ..*request
             })
             .await?;
+        validate_list_page(page.is_truncated, request.continuation_token, page.next_continuation_token.as_deref())?;
         page.objects = page
             .objects
             .into_iter()
@@ -800,11 +805,6 @@ impl SourceBackend for S3SourceBackend {
 
         let is_truncated = output.is_truncated.unwrap_or(false);
         let next_continuation_token = output.next_continuation_token;
-        if is_truncated && next_continuation_token.is_none() {
-            return Err(SourceError::Other(
-                "source reported a truncated listing without a continuation token".to_string(),
-            ));
-        }
         let objects = output
             .contents
             .unwrap_or_default()
@@ -1274,7 +1274,9 @@ mod tests {
 <CommonPrefixes><Prefix>data/photos/</Prefix></CommonPrefixes>
 <CommonPrefixes><Prefix>outside/</Prefix></CommonPrefixes>
 </ListBucketResult>"#;
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), body)]).await;
+        let next_body = body.replace("data/opaque", "data/next");
+        let (client, requests) =
+            scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), &next_body)]).await;
         let first = client
             .list_page(&SourceListRequest {
                 prefix: Some("photos/"),
@@ -1336,7 +1338,104 @@ mod tests {
             .list_objects_v2(None, None, 10)
             .await
             .expect_err("truncated page without token is corrupt");
-        assert!(matches!(err, SourceError::Other(_)), "{err:?}");
+        assert!(matches!(err, SourceError::InvalidPagination(ListPageError::Missing)), "{err:?}");
+    }
+
+    #[tokio::test]
+    async fn list_page_validates_s3_cursor_progress_before_mapping_entries() {
+        for contents in ["", "<Contents><Key>data/a</Key><Size>1</Size></Contents>"] {
+            for (truncated, next, expected) in [
+                (true, None, Some(ListPageError::Missing)),
+                (true, Some(""), Some(ListPageError::Empty)),
+                (true, Some("stuck"), Some(ListPageError::Repeated)),
+                (true, Some("opaque-next"), None),
+                (false, None, None),
+                (false, Some("stuck"), None),
+            ] {
+                let next_xml = next
+                    .map(|next| format!("<NextContinuationToken>{next}</NextContinuationToken>"))
+                    .unwrap_or_default();
+                let body = format!(
+                    "<ListBucketResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><IsTruncated>{truncated}</IsTruncated>{next_xml}{contents}</ListBucketResult>"
+                );
+                let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), &body)]).await;
+                let result = client
+                    .list_page(&SourceListRequest {
+                        continuation_token: Some("stuck"),
+                        max_keys: 2,
+                        ..Default::default()
+                    })
+                    .await;
+                match expected {
+                    Some(expected) => {
+                        let error = result.expect_err("malformed pagination must fail at the provider boundary");
+                        assert!(
+                            matches!(&error, SourceError::InvalidPagination(actual) if *actual == expected),
+                            "{error:?}"
+                        );
+                        assert_eq!(error.class_label(), "invalid_pagination");
+                        assert!(!error.is_retryable());
+                        assert!(!error.to_string().contains("stuck"), "errors must not echo opaque tokens");
+                    }
+                    None => {
+                        let page = result.expect("progressing empty/nonempty pages and EOF are valid");
+                        assert_eq!(page.is_truncated, truncated);
+                        assert_eq!(page.next_continuation_token.as_deref(), next);
+                        assert_eq!(page.objects.len(), usize::from(!contents.is_empty()));
+                        if let Some(object) = page.objects.first() {
+                            assert_eq!(object.key, "a");
+                        }
+                    }
+                }
+                let requests = recorded(&requests);
+                assert_eq!(requests.len(), 1, "invalid pagination must not be retried");
+                assert!(requests[0].uri.contains("continuation-token=stuck"));
+            }
+        }
+    }
+
+    struct ListOnlyBackend(SourcePage);
+
+    #[async_trait::async_trait]
+    impl SourceBackend for ListOnlyBackend {
+        async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
+            assert_eq!(request.continuation_token, Some("stuck"), "opaque cursors reach every provider unchanged");
+            Ok(self.0.clone())
+        }
+
+        async fn head(&self, _key: &str) -> Result<SourceHead, SourceError> {
+            panic!("unexpected HEAD in list test")
+        }
+        async fn get(&self, _key: &str, _range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
+            panic!("unexpected GET in list test")
+        }
+        async fn tagging(&self, _key: &str) -> Result<HashMap<String, String>, SourceError> {
+            panic!("unexpected tagging in list test")
+        }
+        async fn probe(&self) -> Result<(), SourceError> {
+            panic!("unexpected probe in list test")
+        }
+    }
+
+    #[tokio::test]
+    async fn list_page_validates_non_s3_provider_cursors_at_the_common_boundary() {
+        for (next, expected) in [
+            (None, ListPageError::Missing),
+            (Some(""), ListPageError::Empty),
+            (Some("stuck"), ListPageError::Repeated),
+        ] {
+            let mut client = prefix_client(Some("data/".into()));
+            client.backend = Box::new(ListOnlyBackend(SourcePage {
+                is_truncated: true,
+                next_continuation_token: next.map(str::to_string),
+                ..Default::default()
+            }));
+            let error = client
+                .list_objects_v2(None, Some("stuck"), 2)
+                .await
+                .expect_err("all providers must advance pagination");
+            assert!(matches!(error, SourceError::InvalidPagination(actual) if actual == expected));
+        }
     }
 
     const TAGGING_BODY: &str = r#"<?xml version="1.0" encoding="UTF-8"?>

crates/ecstore/src/bucket/on_demand_migration/stats.rs

@@ -177,7 +177,7 @@ impl From<&SourceError> for PullFailureReason {
             SourceError::Connect(_) => PullFailureReason::SourceConnect,
             SourceError::ServerError(_) => PullFailureReason::SourceServerError,
             SourceError::Unsupported(_) => PullFailureReason::SourceUnsupported,
-            SourceError::Other(_) => PullFailureReason::SourceOther,
+            SourceError::InvalidPagination(_) | SourceError::Other(_) => PullFailureReason::SourceOther,
         }
     }
 }

crates/ecstore/src/core/pools.rs

@@ -5493,6 +5493,7 @@ where
     fence.ensure_held()?;
     let mut opts = ObjectOptions {
         max_parity: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         no_lock: true,
         http_preconditions: Some(pool_meta_cas_preconditions(token, object)?),
         ..Default::default()
@@ -14412,6 +14413,7 @@ impl ECStore {
                 encoded.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()
@@ -14566,6 +14568,7 @@ impl ECStore {
                 encoded,
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(http_preconditions),
                     ..Default::default()
                 },
@@ -14957,6 +14960,7 @@ impl ECStore {
                 encoded,
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_match: Some(etag),
                         ..Default::default()

crates/ecstore/src/disk/disk_store.rs

@@ -317,6 +317,22 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
         dst_path: &str,
         external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> Result<RenameDataResp> {
+        self.rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, external_guard)
+            .await
+            .result
+    }
+}
+
+impl LocalDiskWrapper {
+    pub(in crate::disk) async fn rename_data_observed(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> super::RenameDataObservation {
         let operation = self.clone();
         let src_volume = src_volume.to_owned();
         let src_path = src_path.to_owned();
@@ -333,22 +349,35 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
         } else {
             get_max_timeout_duration()
         };
-        run_owned_mutation(external_guard, move || async move {
-            operation
+        let observed = run_owned_mutation(external_guard, move || async move {
+            let mut preflight_rejection = None;
+            let result = operation
                 .track_disk_health_mutation(
                     "rename_data",
                     DiskMetricMutation::Write,
                     || async {
-                        operation
-                            .disk
-                            .rename_data_borrowed(&src_volume, &src_path, &fi, &dst_volume, &dst_path)
-                            .await
+                        // Preserve the former DiskAPI future's single boxing boundary.
+                        let observed =
+                            Box::pin(
+                                operation
+                                    .disk
+                                    .rename_data_observed(&src_volume, &src_path, &fi, &dst_volume, &dst_path),
+                            )
+                            .await;
+                        preflight_rejection = observed.preflight_rejection;
+                        observed.result
                     },
                     timeout_duration,
                 )
-                .await
+                .await;
+            // Health tracking must observe the real disk error, not an Ok tuple.
+            Ok(super::RenameDataObservation {
+                result,
+                preflight_rejection,
+            })
         })
-        .await
+        .await;
+        observed.unwrap_or_else(|error| super::RenameDataObservation::unknown(Err(error)))
     }
 }
 
@@ -2588,6 +2617,46 @@ mod tests {
         assert_eq!(wrapper.metrics_snapshot().api_calls.get("unknown"), Some(&1));
     }
 
+    #[tokio::test]
+    async fn rename_preflight_evidence_preserves_health_errors_and_owned_reply() {
+        for source_exists in [false, true] {
+            for guarded in [false, true] {
+                let dir = tempfile::tempdir().expect("temp dir should be created");
+                let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8"))
+                    .expect("endpoint should parse");
+                let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+                if source_exists {
+                    disk.make_volume("source").await.expect("source volume should exist");
+                }
+                let wrapper = LocalDiskWrapper::new(disk, false);
+                let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+                let external_guard = guarded.then(|| Arc::new(DropProbe(Arc::clone(&drops))) as Arc<dyn Send + Sync>);
+                let mut file_info = FileInfo::new("object", 1, 0);
+                file_info.mod_time = Some(::time::OffsetDateTime::now_utc());
+                file_info.erasure.index = 1;
+                let observed = wrapper
+                    .rename_data_observed("source", "object", &file_info, "missing-destination", "object", external_guard)
+                    .await;
+                assert!(observed.rejected_before_publication(), "normal access rejection must carry proof");
+                assert!(matches!(observed.result, Err(DiskError::VolumeNotFound)));
+                let snapshot = wrapper.metrics_snapshot();
+                assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1));
+                assert_eq!(snapshot.total_writes, 0, "health tracking must not observe the rejection as Ok");
+                assert_eq!(drops.load(Ordering::SeqCst), usize::from(guarded));
+
+                wrapper.health.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+                let observed = wrapper
+                    .rename_data_observed("source", "object", &file_info, "missing-destination", "object", None)
+                    .await;
+                assert!(!observed.rejected_before_publication(), "wrapper errors carry no local preflight proof");
+                assert!(matches!(observed.result, Err(DiskError::FaultyDisk)));
+                let snapshot = wrapper.metrics_snapshot();
+                assert_eq!(snapshot.total_errors_availability, 1);
+                assert_eq!(snapshot.total_writes, 0);
+            }
+        }
+    }
+
     #[tokio::test]
     async fn local_disk_health_wrapper_counts_returned_availability_errors() {
         let dir = tempfile::tempdir().expect("temp dir should be created");

crates/ecstore/src/disk/mod.rs

@@ -75,6 +75,25 @@ use time::OffsetDateTime;
 use tokio::io::{AsyncRead, AsyncWrite};
 use uuid::Uuid;
 
+/// Local preflight evidence stays outside DiskAPI and the RPC response format.
+pub(crate) struct RenameDataObservation {
+    pub(crate) result: Result<RenameDataResp>,
+    preflight_rejection: Option<local::LocalRenamePreflightRejection>,
+}
+
+impl RenameDataObservation {
+    fn unknown(result: Result<RenameDataResp>) -> Self {
+        Self {
+            result,
+            preflight_rejection: None,
+        }
+    }
+
+    pub(crate) fn rejected_before_publication(&self) -> bool {
+        self.result.is_err() && self.preflight_rejection.is_some()
+    }
+}
+
 const QUOTA_MUTATION_FENCE_PREFIX: &str = "tmp/quota-mutation-fences/";
 pub(crate) const QUOTA_MUTATION_FENCE_METADATA_SUFFIX: &str = "quota-mutation-fence-token";
 
@@ -711,6 +730,36 @@ impl Disk {
             .await
     }
 
+    pub(crate) async fn rename_data_borrowed_with_fence_observed(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        scanner_publication_lease_token: Option<Uuid>,
+    ) -> RenameDataObservation {
+        match self {
+            Disk::Local(local_disk) => {
+                local_disk
+                    .rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, None)
+                    .await
+            }
+            Disk::Remote(remote_disk) => RenameDataObservation::unknown(
+                remote_disk
+                    .rename_data_borrowed_with_fence(
+                        src_volume,
+                        src_path,
+                        fi,
+                        dst_volume,
+                        dst_path,
+                        scanner_publication_lease_token,
+                    )
+                    .await,
+            ),
+        }
+    }
+
     pub(crate) async fn rename_data_borrowed_with_fence(
         &self,
         src_volume: &str,

crates/ecstore/src/object_api/types.rs

@@ -870,6 +870,18 @@ impl TierFreeVersionReceiptSink {
     }
 }
 
+/// Internal PUT completion boundary; this does not change fsync or write quorum.
+#[doc(hidden)]
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum WriteCompletion {
+    /// Return at write quorum when the commit owner can retain its guards.
+    #[default]
+    Quorum,
+    /// Drain the rename fan-out before returning. Minority failures still heal
+    /// after a successful quorum commit; this does not require every disk to succeed.
+    TailDrained,
+}
+
 #[derive(Default, Clone)]
 pub struct ObjectOptions {
     // Use the maximum parity (N/2), used when saving server configuration files
@@ -896,6 +908,10 @@ pub struct ObjectOptions {
     /// Persisted bucket incarnation observed before authorization.
     pub expected_bucket_incarnation_id: Option<Uuid>,
     pub no_lock: bool,
+    /// Control-plane writers that immediately read or CAS the same namespace
+    /// key use TailDrained without changing namespace lock ownership.
+    #[doc(hidden)]
+    pub write_completion: WriteCompletion,
     /// True when an upper layer already holds the object read lock before
     /// forwarding a no_lock read to the set layer.
     pub metadata_cache_safe: bool,

crates/ecstore/src/services/tier/tier_mutation_intent.rs

@@ -460,6 +460,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -556,6 +557,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()

crates/ecstore/src/services/tier/tier_probe_intent.rs

@@ -494,6 +494,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -549,6 +550,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current.record_etag.clone()),
                 ..Default::default()

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

@@ -0,0 +1,385 @@
+// 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.
+
+//! Pure metadata quorum and early-stop decisions for `SetDisks` reads.
+//!
+//! Disk scheduling, coalescing, cancellation, and late shard materialization
+//! remain with their existing owners; this module only classifies observations.
+
+use crate::diagnostics::get::{
+    GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER,
+    GET_METADATA_EARLY_STOP_REASON_ERROR, GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM,
+    GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST,
+    GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM,
+    GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND,
+};
+use crate::disk::error::DiskError;
+use crate::disk::error_reduce::OBJECT_OP_IGNORED_ERRS;
+use crate::set_disk::file_info_is_valid_for_metadata;
+use rustfs_filemeta::FileInfo;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(in crate::set_disk) struct MetadataEarlyStopDecision {
+    pub(in crate::set_disk) reason: &'static str,
+}
+
+#[derive(Clone, Debug)]
+pub(in crate::set_disk) struct MetadataQuorumAccumulator {
+    pub(in crate::set_disk) total_disks: usize,
+    pub(in crate::set_disk) default_parity_count: usize,
+    pub(in crate::set_disk) allow_early_stop: bool,
+    pub(in crate::set_disk) valid_responses: usize,
+    pub(in crate::set_disk) not_found_responses: usize,
+    pub(in crate::set_disk) version_not_found_responses: usize,
+    pub(in crate::set_disk) ignored_errors: usize,
+    pub(in crate::set_disk) hard_errors: usize,
+    pub(in crate::set_disk) candidate: Option<FileInfo>,
+    pub(in crate::set_disk) candidate_votes: usize,
+    // Bitset of shard indexes whose metadata matches the candidate. Erasure
+    // layouts are capped at 16 shards, so this stays allocation-free on the
+    // GET metadata hot path.
+    candidate_shard_mask: u16,
+    pub(in crate::set_disk) conflicting_metadata: bool,
+    pub(in crate::set_disk) delete_marker_seen: bool,
+    pub(in crate::set_disk) delete_marker_candidates: Vec<(FileInfo, usize)>,
+    pub(in crate::set_disk) delete_marker_votes: usize,
+    pub(in crate::set_disk) requested_version_id: String,
+    pub(in crate::set_disk) matching_version_votes: usize,
+}
+
+impl MetadataQuorumAccumulator {
+    pub(in crate::set_disk) fn new(total_disks: usize, default_parity_count: usize, allow_early_stop: bool) -> Self {
+        Self {
+            total_disks,
+            default_parity_count,
+            allow_early_stop,
+            valid_responses: 0,
+            not_found_responses: 0,
+            version_not_found_responses: 0,
+            ignored_errors: 0,
+            hard_errors: 0,
+            candidate: None,
+            candidate_votes: 0,
+            candidate_shard_mask: 0,
+            conflicting_metadata: false,
+            delete_marker_seen: false,
+            delete_marker_candidates: Vec::new(),
+            delete_marker_votes: 0,
+            requested_version_id: String::new(),
+            matching_version_votes: 0,
+        }
+    }
+
+    pub(in crate::set_disk) fn with_requested_version_id(mut self, version_id: &str) -> Self {
+        self.requested_version_id = version_id.to_string();
+        self
+    }
+
+    pub(in crate::set_disk) fn observe_file_info(&mut self, file_info: &FileInfo) {
+        self.observe_file_info_with_index(None, file_info);
+    }
+
+    pub(in crate::set_disk) fn observe_file_info_at(&mut self, disk_index: usize, file_info: &FileInfo) {
+        self.observe_file_info_with_index(Some(disk_index), file_info);
+    }
+
+    fn observe_file_info_with_index(&mut self, disk_index: Option<usize>, file_info: &FileInfo) {
+        if !file_info_is_valid_for_metadata(file_info) {
+            self.hard_errors = self.hard_errors.saturating_add(1);
+            return;
+        }
+
+        self.valid_responses = self.valid_responses.saturating_add(1);
+
+        // Track version match for versioned requests
+        if !self.requested_version_id.is_empty()
+            && let Some(ref vid) = file_info.version_id
+            && vid.to_string() == self.requested_version_id
+        {
+            self.matching_version_votes = self.matching_version_votes.saturating_add(1);
+        }
+
+        if file_info.is_canonical_delete_marker() {
+            self.delete_marker_seen = true;
+            if let Some((_, votes)) = self
+                .delete_marker_candidates
+                .iter_mut()
+                .find(|(candidate, _)| metadata_early_stop_candidate_matches(candidate, file_info))
+            {
+                *votes = votes.saturating_add(1);
+            } else {
+                self.delete_marker_candidates.push((file_info.clone(), 1));
+            }
+            self.delete_marker_votes = self
+                .delete_marker_candidates
+                .iter()
+                .map(|(_, votes)| *votes)
+                .max()
+                .unwrap_or_default();
+            self.conflicting_metadata |= self.delete_marker_candidates.len() > 1;
+            return;
+        }
+
+        match &self.candidate {
+            Some(candidate) if metadata_early_stop_candidate_matches(candidate, file_info) => {
+                self.candidate_votes = self.candidate_votes.saturating_add(1);
+                if let Some(disk_index) = disk_index
+                    && let Some(bit) = Self::candidate_shard_bit(candidate, file_info, disk_index)
+                {
+                    self.candidate_shard_mask |= bit;
+                }
+            }
+            Some(_) => {
+                self.conflicting_metadata = true;
+            }
+            None => {
+                self.candidate = Some(file_info.clone());
+                self.candidate_votes = 1;
+                if let Some(disk_index) = disk_index
+                    && let Some(bit) = Self::candidate_shard_bit(file_info, file_info, disk_index)
+                {
+                    self.candidate_shard_mask |= bit;
+                }
+            }
+        }
+    }
+
+    fn candidate_shard_bit(candidate: &FileInfo, file_info: &FileInfo, disk_index: usize) -> Option<u16> {
+        let &erasure_index = candidate.erasure.distribution.get(disk_index)?;
+        if erasure_index == 0 || erasure_index > u16::BITS as usize || file_info.erasure.index != erasure_index {
+            return None;
+        }
+        Some(1u16 << (erasure_index - 1))
+    }
+
+    pub(in crate::set_disk) fn candidate_has_read_reserve(&self) -> bool {
+        self.candidate_read_reserve_target()
+            .is_some_and(|required| self.candidate_shard_mask.count_ones() as usize >= required)
+    }
+
+    pub(in crate::set_disk) fn candidate_read_reserve_target(&self) -> Option<usize> {
+        let candidate = self.candidate.as_ref()?;
+        Some(
+            candidate
+                .erasure
+                .data_blocks
+                .saturating_add(usize::from(candidate.erasure.parity_blocks > 0)),
+        )
+    }
+
+    pub(in crate::set_disk) fn observe_error(&mut self, err: &DiskError) {
+        match err {
+            DiskError::FileNotFound | DiskError::VolumeNotFound => {
+                self.not_found_responses = self.not_found_responses.saturating_add(1);
+            }
+            DiskError::FileVersionNotFound => {
+                self.version_not_found_responses = self.version_not_found_responses.saturating_add(1);
+            }
+            _ if is_metadata_fanout_ignored_error(err) => {
+                self.ignored_errors = self.ignored_errors.saturating_add(1);
+            }
+            _ => {
+                self.hard_errors = self.hard_errors.saturating_add(1);
+            }
+        }
+    }
+
+    pub(in crate::set_disk) fn early_stop_decision(&self) -> Option<MetadataEarlyStopDecision> {
+        if !self.allow_early_stop {
+            return None;
+        }
+        if self.delete_marker_votes >= self.default_write_quorum() {
+            return Some(MetadataEarlyStopDecision {
+                reason: GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER,
+            });
+        }
+        if self.conflicting_metadata
+            || self.delete_marker_seen
+            || self.not_found_responses > 0
+            || self.version_not_found_responses > 0
+            || self.hard_errors > 0
+        {
+            return None;
+        }
+        if self
+            .candidate
+            .as_ref()
+            .and_then(|candidate| self.candidate_latest_quorum(candidate))
+            .is_some_and(|latest_quorum| self.candidate_votes >= latest_quorum)
+        {
+            return Some(MetadataEarlyStopDecision {
+                reason: GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM,
+            });
+        }
+        None
+    }
+
+    /// Check if a versioned request can early-stop because the requested
+    /// version_id has reached quorum across disks.
+    pub(in crate::set_disk) fn version_early_stop_decision(&self) -> Option<MetadataEarlyStopDecision> {
+        if !self.allow_early_stop {
+            return None;
+        }
+        if self.requested_version_id.is_empty() {
+            return None;
+        }
+        if self.conflicting_metadata
+            || self.delete_marker_seen
+            || self.not_found_responses > 0
+            || self.version_not_found_responses > 0
+            || self.hard_errors > 0
+        {
+            return None;
+        }
+        if self.matching_version_votes >= self.read_quorum_for_version() {
+            return Some(MetadataEarlyStopDecision {
+                reason: GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM,
+            });
+        }
+        None
+    }
+
+    pub(in crate::set_disk) fn can_still_reach_early_stop_with_pending(&self, pending: usize) -> bool {
+        if !self.allow_early_stop {
+            return false;
+        }
+        if self.delete_marker_votes.saturating_add(pending) >= self.default_write_quorum() {
+            return true;
+        }
+        if self.conflicting_metadata
+            || self.delete_marker_seen
+            || self.not_found_responses > 0
+            || self.version_not_found_responses > 0
+            || self.hard_errors > 0
+        {
+            return false;
+        }
+        if !self.requested_version_id.is_empty()
+            && self.matching_version_votes.saturating_add(pending) >= self.read_quorum_for_version()
+        {
+            return true;
+        }
+        match &self.candidate {
+            Some(candidate) => self
+                .candidate_latest_quorum(candidate)
+                .is_some_and(|latest_quorum| self.candidate_votes.saturating_add(pending) >= latest_quorum),
+            None => pending >= self.default_write_quorum(),
+        }
+    }
+
+    /// Compute the read quorum threshold for version-aware early-stop.
+    /// Uses `total_disks / 2` (like `missing_response_quorum`) when
+    /// `default_parity_count` is set, otherwise requires all disks.
+    pub(in crate::set_disk) fn read_quorum_for_version(&self) -> usize {
+        self.missing_response_quorum()
+    }
+
+    pub(in crate::set_disk) fn final_miss_reason(&self) -> &'static str {
+        if !self.allow_early_stop {
+            return GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST;
+        }
+        if self.conflicting_metadata {
+            return GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA;
+        }
+        if self.delete_marker_seen {
+            return GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER;
+        }
+        let missing_response_quorum = self.missing_response_quorum();
+        if self.version_not_found_responses >= missing_response_quorum {
+            return GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND;
+        }
+        if self.not_found_responses >= missing_response_quorum {
+            return GET_METADATA_EARLY_STOP_REASON_NOT_FOUND;
+        }
+        if self.hard_errors > 0 {
+            return GET_METADATA_EARLY_STOP_REASON_ERROR;
+        }
+        if self.ignored_errors > 0 {
+            return GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM;
+        }
+        GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM
+    }
+
+    pub(in crate::set_disk) fn candidate_latest_quorum(&self, candidate: &FileInfo) -> Option<usize> {
+        if self.default_parity_count == 0 {
+            return Some(self.total_disks);
+        }
+        if candidate.is_canonical_delete_marker() || candidate.size == 0 || candidate.erasure.parity_blocks >= self.total_disks {
+            return None;
+        }
+        let data_blocks = candidate.erasure.data_blocks;
+        Some(if data_blocks == candidate.erasure.parity_blocks {
+            data_blocks.saturating_add(1)
+        } else {
+            data_blocks
+        })
+    }
+
+    pub(crate) fn default_write_quorum(&self) -> usize {
+        if self.default_parity_count == 0 || self.default_parity_count >= self.total_disks {
+            return self.total_disks;
+        }
+        let data_blocks = self.total_disks.saturating_sub(self.default_parity_count);
+        if data_blocks == self.default_parity_count {
+            data_blocks.saturating_add(1)
+        } else {
+            data_blocks
+        }
+    }
+
+    pub(in crate::set_disk) fn missing_response_quorum(&self) -> usize {
+        if self.default_parity_count == 0 || self.default_parity_count >= self.total_disks {
+            self.total_disks
+        } else {
+            self.total_disks / 2
+        }
+    }
+}
+
+pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo, right: &FileInfo) -> bool {
+    left.volume == right.volume
+        && left.name == right.name
+        && left.version_id == right.version_id
+        && left.is_latest == right.is_latest
+        && left.deleted == right.deleted
+        && left.mark_deleted == right.mark_deleted
+        && left.transition_status == right.transition_status
+        && left.transitioned_objname == right.transitioned_objname
+        && left.transition_tier == right.transition_tier
+        && left.transition_version_id == right.transition_version_id
+        && left.transition_version == right.transition_version
+        && left.transition_version_state == right.transition_version_state
+        && left.expire_restored == right.expire_restored
+        && left.size == right.size
+        && left.mod_time == right.mod_time
+        && left.mode == right.mode
+        && left.written_by_version == right.written_by_version
+        && left.metadata == right.metadata
+        && left.replication_state_internal == right.replication_state_internal
+        && left.parts == right.parts
+        && left.checksum == right.checksum
+        && left.versioned == right.versioned
+        && left.num_versions == right.num_versions
+        && left.successor_mod_time == right.successor_mod_time
+        && left.data_dir == right.data_dir
+        && left.erasure.algorithm == right.erasure.algorithm
+        && left.erasure.data_blocks == right.erasure.data_blocks
+        && left.erasure.parity_blocks == right.erasure.parity_blocks
+        && left.erasure.block_size == right.erasure.block_size
+        && left.erasure.distribution == right.erasure.distribution
+}
+
+pub(in crate::set_disk) fn is_metadata_fanout_ignored_error(err: &DiskError) -> bool {
+    OBJECT_OP_IGNORED_ERRS.iter().any(|ignored| ignored == err)
+}

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

@@ -18,3 +18,4 @@
 //! duplicating read/write/erasure logic.
 
 pub(crate) mod io_primitives;
+mod metadata_quorum;

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

@@ -299,11 +299,11 @@ use crate::error::is_err_invalid_upload_id;
 use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed};
 use crate::object_api::{
     NamespaceLockFence, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode,
-    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY,
+    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, WriteCompletion,
 };
 use crate::services::notification_sys::RemoteVersionStateFleetProofToken;
 use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata};
-use crate::set_disk::core::io_primitives::{RenameTailCleanup, finish_rename_tail_heal};
+use crate::set_disk::core::io_primitives::{RenameRollbackReceipt, RenameTailCleanup, finish_rename_tail_heal};
 #[cfg(test)]
 use crate::storage_api_contracts::namespace::NamespaceLocking;
 #[cfg(test)]
@@ -3548,6 +3548,7 @@ impl SetDisks {
                 (None, None, None)
             };
         let mut tmp_cleanup_owned = false;
+        let rollback_receipt = RenameRollbackReceipt::default();
         let operation = async {
             let erasure = Arc::new(erasure_from_file_info(&fi, false)?);
 
@@ -4256,6 +4257,7 @@ impl SetDisks {
             let commit_bucket = bucket.to_owned();
             let commit_object = object.to_owned();
             let commit_tmp_dir = tmp_dir.clone();
+            let commit_rollback_receipt = rollback_receipt.clone();
             let commit_object_lock_guard = object_lock_guard.take();
             let commit_decommission_object_lock_guard = decommission_object_lock_guard.take();
             let commit_publication_guard = publication_commit_guard.take();
@@ -4266,13 +4268,17 @@ impl SetDisks {
             // complete rename fan-out drains. Keep this path synchronous so
             // its terminal state is known before the coordinator releases
             // remote leases.
-            let commit_allows_early_ack = !(opts.data_movement && opts.has_decommission_capacity_reservation())
-                && (commit_object_lock_guard.is_some()
-                    || commit_decommission_object_lock_guard.is_some()
-                    || commit_publication_guard.is_some())
+            let commit_owns_namespace_guard = commit_object_lock_guard.is_some()
+                || commit_decommission_object_lock_guard.is_some()
+                || commit_publication_guard.is_some();
+            let commit_allows_early_ack = opts.write_completion == WriteCompletion::Quorum
+                && !(opts.data_movement && opts.has_decommission_capacity_reservation())
+                && commit_owns_namespace_guard
                 && commit_scanner_publication_scope.is_none();
+            // Full-tail callers also transfer owned guards to the coordinator:
+            // cancelling their ACK waiter must not cancel an in-flight rename.
             let detach_commit_owner = commit_scanner_publication_scope.is_some()
-                || commit_allows_early_ack
+                || commit_owns_namespace_guard
                 || commit_bucket_lifecycle_guard.is_some()
                 || quota_mutation_fence;
             let commit_write_path_label = write_path.metric_label();
@@ -4452,7 +4458,8 @@ impl SetDisks {
                         write_quorum,
                         commit_scanner_publication_lease_tokens.as_ref(),
                     )
-                    .with_publication_scope(commit_scanner_publication_scope.clone()),
+                    .with_publication_scope(commit_scanner_publication_scope.clone())
+                    .with_rollback_receipt(commit_rollback_receipt.clone()),
                 )
                 .await;
                 if let Some(scope) = commit_scanner_publication_scope.as_ref() {
@@ -4585,6 +4592,11 @@ impl SetDisks {
                 let rename_commit = match rename_result {
                     Ok(commit) => commit,
                     Err(err) => {
+                        if commit_rollback_receipt.is_incomplete() {
+                            // Incomplete undo retains the staging source and
+                            // rollback backup for recovery; cleanup is unsafe.
+                            return Err(err.into());
+                        }
                         if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
                             warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data");
                         } else if issue3031_diag_enabled() {
@@ -4617,9 +4629,8 @@ impl SetDisks {
                     request.object_version_id = committed_version_id
                         .or_else(|| commit_version_suspended.then(Uuid::nil))
                         .map(|version_id| version_id.to_string());
-                    tokio::spawn(async move {
-                        let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await;
-                    });
+                    let heal_set = commit_set.clone();
+                    tokio::spawn(async move { heal_set.submit_rename_tail_heal(request).await });
                 }
 
                 let rename_stage_elapsed = rename_stage_start.elapsed();
@@ -4885,7 +4896,7 @@ impl SetDisks {
                     );
                 }
             });
-        } else {
+        } else if !rollback_receipt.is_incomplete() {
             // Failure path (quorum loss / rollback): keep the cleanup inline so
             // a failed PUT never returns while its tmp shards are still on disk
             // (state-residue hardening tracked by backlog#864 / backlog#898).
@@ -17494,27 +17505,69 @@ mod put_object_tmp_cleanup_tests {
     }
 
     #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
     async fn put_object_failure_cleans_tmp_workspace_inline() {
-        let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
+                let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
+                let bucket = "tmp-clean-missing-bucket";
+                let object = "orphan-object";
+                let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
+                let writer = Arc::clone(&set_disks);
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
+                    writer
+                        .put_object(
+                            bucket,
+                            object,
+                            &mut reader,
+                            &ObjectOptions {
+                                write_completion,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                });
+                tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                    .await
+                    .expect("missing-bucket PUT must stage before rename");
+                let staged = non_trash_tmp_entries(&temp_dirs).await;
+                assert_eq!(staged.len(), 4, "every disk must have a staged workspace before rejection");
+                for workspace in staged {
+                    let mut entries = tokio::fs::read_dir(&workspace)
+                        .await
+                        .expect("staged workspace should be readable");
+                    let mut shards = 0;
+                    while let Some(entry) = entries.next_entry().await.expect("staged data directory should be readable") {
+                        if entry.file_type().await.expect("staged entry type").is_dir() {
+                            let part = tokio::fs::metadata(entry.path().join("part.1"))
+                                .await
+                                .expect("staging must contain an actual erasure shard");
+                            assert!(part.len() > 0, "the shard must be written before the missing-bucket failure");
+                            shards += 1;
+                        }
+                    }
+                    assert_eq!(shards, 1);
+                }
+                assert!(temp_dirs.iter().all(|dir| !dir.path().join(bucket).exists()));
+                barrier.release();
+                let err = tokio::time::timeout(Duration::from_secs(30), put)
+                    .await
+                    .expect("missing-bucket PUT must finish")
+                    .expect("PUT task should join")
+                    .expect_err("put_object into a missing bucket volume must fail");
+                assert!(matches!(err, StorageError::VolumeNotFound), "original disk error expected: {err}");
 
-        // The bucket volume is never created, so the shards are written into
-        // the tmp workspace and the commit fails at rename_data with a quorum
-        // error — exercising the failure-path cleanup.
-        let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
-        let err = set_disks
-            .put_object("tmp-clean-missing-bucket", "orphan-object", &mut reader, &ObjectOptions::default())
-            .await
-            .expect_err("put_object into a missing bucket volume must fail");
-
-        // No polling: the failure path must clean the tmp workspace inline,
-        // before put_object returns (backlog#864 / backlog#898 hardening).
-        let leftovers = non_trash_tmp_entries(&temp_dirs).await;
-        assert!(
-            leftovers.is_empty(),
-            "failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
-        );
-
-        drop(temp_dirs);
+                // No polling: known pre-publication rejection must clean staging
+                // inline, before PUT returns (backlog#864 / backlog#898).
+                let leftovers = non_trash_tmp_entries(&temp_dirs).await;
+                assert!(
+                    leftovers.is_empty(),
+                    "failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
+                );
+            }
+        })
+        .await;
     }
 
     #[tokio::test]
@@ -18157,6 +18210,354 @@ mod put_object_tmp_cleanup_tests {
         .await;
     }
 
+    async fn make_completion_test_bucket(disks: &[DiskStore], bucket: &str) {
+        for disk in disks {
+            disk.make_volume(bucket)
+                .await
+                .expect("completion test bucket should be created");
+        }
+    }
+
+    /// Observe the actual metadata quorum while the remaining rename is parked.
+    /// A completed task count alone can race tasks that have not started yet.
+    async fn wait_for_paused_tail_metadata_quorum(disks: &[DiskStore], bucket: &str, object: &str) {
+        tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                let mut committed = 0;
+                for disk in disks {
+                    match disk.read_version("", bucket, object, "", &ReadOptions::default()).await {
+                        Ok(_) => committed += 1,
+                        Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {}
+                        Err(err) => panic!("unexpected metadata error while observing {bucket}/{object}: {err}"),
+                    }
+                }
+                if committed == 3 {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("three disks must publish metadata while the fourth rename remains paused");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_waits_for_tail_and_allows_immediate_cas() {
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for size in [4096, 1024 * 1024] {
+                let (_dirs, disks, set) = hermetic_set_disks(4).await;
+                let bucket = "put-full-tail-cas";
+                let object = "full-tail-cas-object";
+                make_completion_test_bucket(&disks, bucket).await;
+                let tasks = rename_fanout_barrier::observe_tasks(object);
+                let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+                let writer = Arc::clone(&set);
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(vec![b'1'; size]);
+                    writer
+                        .put_object(
+                            bucket,
+                            object,
+                            &mut reader,
+                            &ObjectOptions {
+                                write_completion: WriteCompletion::TailDrained,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                });
+                tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                    .await
+                    .expect("full-tail PUT must reach the rename barrier");
+                wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+                assert!(!put.is_finished(), "full-tail PUT must remain pending after metadata quorum");
+                let mut lock_probe = Box::pin(set.acquire_write_lock_diag("full_tail_probe", bucket, object));
+                assert!(
+                    futures::poll!(lock_probe.as_mut()).is_pending(),
+                    "the owned namespace guard must remain held"
+                );
+                barrier.release();
+                let written = tokio::time::timeout(Duration::from_secs(30), put)
+                    .await
+                    .expect("full-tail PUT should finish after release")
+                    .expect("full-tail PUT task should join")
+                    .expect("full-tail PUT must commit");
+                assert_eq!(tasks.running(), 0, "full-tail response must follow every rename task");
+                drop(
+                    tokio::time::timeout(Duration::from_secs(5), lock_probe)
+                        .await
+                        .expect("same-key lock should be available on return")
+                        .expect("same-key lock probe should succeed"),
+                );
+                for disk in &disks {
+                    disk.read_version("", bucket, object, "", &ReadOptions::default())
+                        .await
+                        .expect("successful full-tail PUT must publish on every healthy disk");
+                }
+                drop(barrier);
+                let mut replacement = PutObjReader::from_vec(b"cas successor".to_vec());
+                set.put_object(
+                    bucket,
+                    object,
+                    &mut replacement,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        http_preconditions: Some(HTTPPreconditions {
+                            if_match: written.etag,
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("immediate same-key CAS must acquire the namespace guard");
+                let mut read = set
+                    .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("CAS successor must be immediately readable");
+                let mut body = Vec::new();
+                read.stream.read_to_end(&mut body).await.expect("successor body must drain");
+                assert_eq!(body, b"cas successor");
+            }
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_preserves_quorum_success_and_heals_failed_tail() {
+        let (_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "put-full-tail-heal";
+        let object = "full-tail-heal-object";
+        make_completion_test_bucket(&disks, bucket).await;
+        let mut heals = set.capture_test_rename_tail_heals();
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let _fault = rename_fault_injection::fail_rename_on(object, &[0]);
+        let writer = Arc::clone(&set);
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+            writer
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("failed tail must first reach the rename barrier");
+        wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+        assert!(!put.is_finished(), "committed quorum must still wait for the failing tail");
+        barrier.release();
+        tokio::time::timeout(Duration::from_secs(30), put)
+            .await
+            .expect("failed tail should drain")
+            .expect("PUT task should join")
+            .expect("a minority tail error must not negate committed quorum");
+        assert_eq!(tasks.running(), 0);
+        let heal = tokio::time::timeout(Duration::from_secs(30), heals.recv())
+            .await
+            .expect("failed tail must schedule heal")
+            .expect("heal capture must remain connected");
+        assert_eq!(heal.bucket, bucket);
+        assert_eq!(heal.object_prefix.as_deref(), Some(object));
+        let info = set
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("committed object must remain readable despite the failed tail");
+        assert_eq!(info.size, TEST_OBJECT_SIZE as i64);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_rejects_quorum_minus_one() {
+        let (_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "put-full-tail-no-quorum";
+        let object = "full-tail-no-quorum-object";
+        make_completion_test_bucket(&disks, bucket).await;
+        let _fault = rename_fault_injection::fail_rename_on(object, &[0, 1]);
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+        let err = set
+            .put_object(
+                bucket,
+                object,
+                &mut reader,
+                &ObjectOptions {
+                    write_completion: WriteCompletion::TailDrained,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("draining two successful disks cannot satisfy write quorum three");
+        assert!(
+            matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
+            "original quorum error expected: {err}"
+        );
+        assert_eq!(tasks.running(), 0, "failed fan-out and rollback must complete before return");
+        assert!(
+            set.get_object_info(bucket, object, &ObjectOptions::default()).await.is_err(),
+            "failed fresh write must not become visible"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn put_incomplete_rollback_preserves_staging_and_old_version_backup() {
+        use crate::set_disk::core::io_primitives::rollback_fault_injection;
+
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
+                for fault in [
+                    rollback_fault_injection::Fault::Io,
+                    rollback_fault_injection::Fault::VolumeNotFoundAfterRename,
+                ] {
+                    let (dirs, disks, set) = hermetic_set_disks(4).await;
+                    let bucket = "put-incomplete-undo";
+                    let object = "incomplete-undo-object";
+                    make_completion_test_bucket(&disks, bucket).await;
+                    let mut old_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]);
+                    set.put_object(
+                        bucket,
+                        object,
+                        &mut old_reader,
+                        &ObjectOptions {
+                            write_completion: WriteCompletion::TailDrained,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("old generation should be completely committed");
+                    wait_for_tmp_workspace_to_drain(&dirs, "old PUT must leave no unrelated staging").await;
+                    let old = disks[0]
+                        .read_version("", bucket, object, "", &ReadOptions::default())
+                        .await
+                        .expect("old metadata must be readable");
+                    let old_data_dir = old.data_dir.expect("non-inline old version needs a data directory");
+                    let tasks = rename_fanout_barrier::observe_tasks(object);
+                    let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+                    let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]);
+                    let _undo_fault = rollback_fault_injection::arm(object, 0, fault);
+                    let writer = Arc::clone(&set);
+                    let put = tokio::spawn(async move {
+                        let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+                        writer
+                            .put_object(
+                                bucket,
+                                object,
+                                &mut reader,
+                                &ObjectOptions {
+                                    write_completion,
+                                    ..Default::default()
+                                },
+                            )
+                            .await
+                    });
+                    tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                        .await
+                        .expect("overwrite must enter the actual rename fan-out before failure injection");
+                    barrier.release();
+                    let err = tokio::time::timeout(Duration::from_secs(30), put)
+                        .await
+                        .expect("incomplete undo must return without hanging")
+                        .expect("PUT task should join")
+                        .expect_err("two renamed disks cannot satisfy write quorum three");
+                    assert!(
+                        matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
+                        "original quorum error expected: {err}"
+                    );
+                    assert_eq!(tasks.running(), 0, "every rename and undo task must be reaped before return");
+                    let leftovers = non_trash_tmp_entries(&dirs).await;
+                    assert!(!leftovers.is_empty(), "incomplete undo must retain the new staging source for recovery");
+                    let backups = dirs
+                        .iter()
+                        .filter(|dir| {
+                            dir.path()
+                                .join(bucket)
+                                .join(object)
+                                .join(old_data_dir.to_string())
+                                .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP)
+                                .exists()
+                        })
+                        .count();
+                    assert_eq!(backups, 1, "exactly the failed undo disk must retain its old-version backup");
+                    // The remaining three disks still serve the old generation;
+                    // the failed minority must never become an acknowledged write.
+                    let mut read = set
+                        .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                        .await
+                        .expect("old generation must remain readable after incomplete rollback");
+                    let mut body = Vec::new();
+                    read.stream
+                        .read_to_end(&mut body)
+                        .await
+                        .expect("old generation should stream");
+                    assert_eq!(body, vec![b'0'; TEST_OBJECT_SIZE]);
+                }
+            }
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_owned_commit_survives_waiter_cancellation() {
+        let (dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = RUSTFS_META_BUCKET;
+        let object = "full-tail-cancelled-receipt";
+        // Internal config writes do not own a bucket lifecycle guard. The object
+        // guard alone must keep the full-tail coordinator alive after cancellation.
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let writer = Arc::clone(&set);
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+            writer
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("cancelled receipt must first reach the rename barrier");
+        wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+        put.abort();
+        assert!(put.await.expect_err("ACK waiter should cancel").is_cancelled());
+        let mut lock_probe = Box::pin(set.acquire_write_lock_diag("cancelled_full_tail_probe", bucket, object));
+        assert!(
+            futures::poll!(lock_probe.as_mut()).is_pending(),
+            "owned coordinator must retain the namespace guard after waiter cancellation"
+        );
+        barrier.release();
+        drop(
+            tokio::time::timeout(Duration::from_secs(30), lock_probe)
+                .await
+                .expect("cancelled coordinator must eventually release its guard")
+                .expect("post-commit lock probe should succeed"),
+        );
+        assert_eq!(tasks.running(), 0, "cancelled coordinator must reap every rename task");
+        for disk in &disks {
+            disk.read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .expect("caller cancellation must not interrupt committed receipt materialization");
+        }
+        wait_for_tmp_workspace_to_drain(&dirs, "cancelled full-tail commit should release staging ownership").await;
+    }
+
     #[tokio::test]
     #[serial_test::serial(capacity_dirty_scope)]
     async fn no_lock_put_waits_for_rename_tail_under_outer_guard() {
@@ -18184,6 +18585,7 @@ mod put_object_tmp_cleanup_tests {
                         &mut reader,
                         &ObjectOptions {
                             no_lock: true,
+                            write_completion: WriteCompletion::TailDrained,
                             ..Default::default()
                         },
                     )
@@ -18209,7 +18611,18 @@ mod put_object_tmp_cleanup_tests {
             put.await
                 .expect("no-lock PUT task should join")
                 .expect("no-lock PUT should commit after the rename tail releases");
+            let mut lock_probe = Box::pin(set_disks.acquire_write_lock_diag("borrowed_full_tail_probe", bucket, object));
+            assert!(
+                futures::poll!(lock_probe.as_mut()).is_pending(),
+                "full-tail PUT must not release the caller's outer guard"
+            );
             drop(outer_guard);
+            drop(
+                tokio::time::timeout(Duration::from_secs(5), lock_probe)
+                    .await
+                    .expect("outer owner releasing its guard should unblock the probe")
+                    .expect("post-outer-guard probe should succeed"),
+            );
         })
         .await;
     }

crates/ecstore/src/set_disk/transition_matrix_tests.rs

@@ -18,6 +18,7 @@ use super::{
 };
 use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, expected_expiry_time};
 use crate::ecstore_validation_blackbox::make_local_set_disks;
+use crate::object_api::WriteCompletion;
 use crate::services::tier::test_util::register_mock_tier;
 use crate::storage_api_contracts::bucket::BucketOperations;
 use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
@@ -72,7 +73,7 @@ async fn transition_and_restore_reclaim_prior_metadata_generations() {
             object,
             &mut reader,
             &ObjectOptions {
-                no_lock: true,
+                write_completion: WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )
@@ -185,7 +186,7 @@ async fn prepared_snapshot_transition_duplicate_and_late_get_use_committed_remot
             object,
             &mut reader,
             &ObjectOptions {
-                no_lock: true,
+                write_completion: WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )

crates/ecstore/src/store/init.rs

@@ -2979,6 +2979,33 @@ mod tests {
     #[cfg(feature = "test-util")]
     const DECOMMISSION_TEST_FAULT_STAGE_TIERED: &str = "decommission_tiered_object";
 
+    fn decommission_retry_fault_hook(
+        bucket: &str,
+        object: &str,
+        faults: Arc<AtomicUsize>,
+    ) -> crate::core::pools::DecommissionTestFaultDecision {
+        let target_bucket = bucket.to_string();
+        let target_object = object.to_string();
+        Arc::new(move |stage, bucket, object, _attempt, succeeded| {
+            if !succeeded
+                || stage != DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT
+                || bucket != target_bucket
+                || object != target_object
+            {
+                return false;
+            }
+
+            // Entry retries reset the local attempt; real copy errors can skip
+            // successful attempts. Only injected faults spend this global budget.
+            faults
+                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
+                    (faults < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS.saturating_sub(1))
+                        .then_some(faults.saturating_add(1))
+                })
+                .is_ok()
+        })
+    }
+
     async fn seed_decommission_source(
         store: &Arc<crate::store::ECStore>,
         bucket: &str,
@@ -5120,6 +5147,33 @@ mod tests {
         shutdown.cancel();
     }
 
+    #[test]
+    fn decommission_retry_fault_budget_counts_successes_across_attempt_changes() {
+        for attempts in [[1, 2, 3], [1, 1, 2], [1, 3, 3]] {
+            let faults = Arc::new(AtomicUsize::new(0));
+            let hook = decommission_retry_fault_hook("bucket", "object", Arc::clone(&faults));
+
+            for (stage, bucket, object, succeeded) in [
+                ("other-stage", "bucket", "object", true),
+                (DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "other-bucket", "object", true),
+                (DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "bucket", "other-object", true),
+                (DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "bucket", "object", false),
+            ] {
+                assert!(!hook(stage, bucket, object, 1, succeeded));
+            }
+            assert_eq!(faults.load(Ordering::SeqCst), 0, "unrelated or failed copies must not consume faults");
+
+            for (index, attempt) in attempts.into_iter().enumerate() {
+                assert_eq!(
+                    hook(DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "bucket", "object", attempt, true),
+                    index < 2,
+                    "attempts={attempts:?}, index={index}"
+                );
+            }
+            assert_eq!(faults.load(Ordering::SeqCst), 2, "attempts={attempts:?}");
+        }
+    }
+
     #[test]
     #[serial_test::serial(storage_class_env)]
     fn decommission_entry_retries_source_changed_without_canceling_other_bucket() {
@@ -5214,31 +5268,8 @@ mod tests {
                     ));
 
                     let ordinary_faults = Arc::new(AtomicUsize::new(0));
-                    let ordinary_faults_for_hook = Arc::clone(&ordinary_faults);
-                    let fault_bucket = other_bucket.clone();
-                    let _fault_guard = crate::core::pools::DecommissionTestFaultGuard::install(Arc::new(
-                        move |stage, bucket, object, attempt, succeeded| {
-                            let candidate = succeeded
-                                && stage == DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT
-                                && bucket == fault_bucket.as_str()
-                                && object == other_object;
-                            if !candidate {
-                                return false;
-                            }
-
-                            // Keep the fault budget global across any
-                            // entry-level re-list; its inner attempt counter
-                            // restarts after SourceChanged.
-                            ordinary_faults_for_hook
-                                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
-                                    let next_fault = faults.saturating_add(1);
-                                    (faults < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS.saturating_sub(1)
-                                        && attempt == next_fault)
-                                        .then_some(next_fault)
-                                })
-                                .is_ok()
-                        },
-                    ));
+                    let fault_hook = decommission_retry_fault_hook(&other_bucket, other_object, Arc::clone(&ordinary_faults));
+                    let _fault_guard = crate::core::pools::DecommissionTestFaultGuard::install(fault_hook);
 
                     let rx = CancellationToken::new();
                     let source_changed_exhaustions = Arc::new(AtomicUsize::new(0));
@@ -8045,10 +8076,15 @@ mod tests {
         );
         assert!(com::read_config(store.pools[0].clone(), &second_page_path).await.is_ok());
 
-        com::save_config(store.pools[target_pool_idx].clone(), &second_page_path, receipt_bytes.clone())
+        let full_tail = ObjectOptions {
+            max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
+            ..Default::default()
+        };
+        com::save_config_with_opts(store.pools[target_pool_idx].clone(), &second_page_path, receipt_bytes.clone(), &full_tail)
             .await
             .expect("second page receipt should restore");
-        com::save_config(store.pools[target_pool_idx].clone(), &second_page_path, b"{corrupt".to_vec())
+        com::save_config_with_opts(store.pools[target_pool_idx].clone(), &second_page_path, b"{corrupt".to_vec(), &full_tail)
             .await
             .expect("second page receipt should corrupt deterministically");
         let corrupt = store

docs/testing/ci-gates.md

@@ -109,3 +109,13 @@ Use an exact preview tag for an end-to-end release rehearsal. Manual dispatches
 ## Change checklist
 
 Update this file in the same PR when a job or check name changes, a workflow gains or loses a `pull_request` or `schedule` trigger, required contexts or strict/merge-queue policy change, report-only vs gating semantics change, or `.github/scheduled-validations.json` membership changes. Do not copy timeouts, crons, or test counts here.
+
+## ECStore invariant selection
+
+The existing `ci.yml` test-and-lint job runs the ordinary ECStore and filemeta tests. After that run, `scripts/check_test_wiring.py --check-core` checks the same nextest profile and package selection against `.config/ecstore-required-tests.json`. Every named test must exist, match the filter, and be non-ignored; the job also requires a nonempty JUnit report. This checks membership without running the tests twice. `core-test-listing.json`, JUnit, and the run log are retained in the existing test-and-lint artifact.
+
+The manifest records a minimum set of invariants: write quorum, metadata rollback, stale-writer lock loss, plaintext Range content, multipart cancellation, hiding uncommitted LIST versions, real MinIO metadata, and corrupt part arrays. Renaming or moving a required test must update the manifest in the same change after checking the compiled listing. Extend this list as new deterministic regressions land; it is not a claim that all storage invariants are covered.
+
+The checked-in MinIO corpus is pinned by file SHA256 and its documented source release. The static wiring guard and the CI selection check both reject missing or changed fixtures. These are metadata fixtures, not a legacy shard-body corpus or proof of crash durability. Optional `legacy_bitrot_read_test` runs may still skip when their external corpus is absent; they do not satisfy a required compatibility lane. Real encrypted fixture reads remain in `minio-interop.yml`, and multi-node fault schedules remain in the existing nightly cluster lane. In-process reopen tests do not establish power-loss durability.
+
+Run `python3 scripts/check_test_wiring.py --self-test` to exercise the negative cases: removed/ignored/filtered tests, malformed listing, absent fixtures, and wrong fixture hashes. Do not update hashes merely to silence the guard; a fixture change needs source/provenance and compatibility review.

docs/testing/ecstore-validation-suite-design.md

@@ -54,6 +54,22 @@ Fail-closed invariants every row enforces:
 
 Fault injection is explicit and deterministic: local disk mocks for unit tests, process-level disk manipulation (`crates/e2e_test/src/chaos.rs`) for e2e tests. Property tests replay a fixed seed for payload, range, and missing-shard selection.
 
+### PUT completion fixtures
+
+`ObjectOptions::default()` uses `WriteCompletion::Quorum`: a namespace-lock-owning PUT may acknowledge write quorum while its rename tail retains the lock. A fixture that immediately inspects every disk or primes a metadata generation must set `write_completion: WriteCompletion::TailDrained` and keep normal locking. TailDrained waits for the existing rename fan-out; it does not require every disk to succeed or change fsync policy. Codec-only `no_lock` fixtures do not cover namespace locking.
+
+The object tests reuse `rename_fanout_barrier::arm(object, disk_slot, phase)` and `observe_tasks(object)`. Wait for the barrier with a deadline, observe actual metadata quorum with `wait_for_paused_tail_metadata_quorum`, then release or cancel. The metadata check distinguishes a real quorum from disk tasks that have not started. Assert zero remaining rename tasks after the owned coordinator releases its lock; cancellation tests also wait for staging cleanup.
+
+| Fixture | Completion boundary |
+|---|---|
+| `early_ack_tail_drain_retains_namespace_lock_until_background_rename_finishes` | Default PUT returns before the parked tail; a second writer remains blocked. |
+| `tail_drained_put_*` | Explicit full-tail PUT retains its guard, preserves quorum success with a failed minority, rejects quorum-minus-one, and survives ACK waiter cancellation. |
+| `transition_and_restore_reclaim_prior_metadata_generations` | Both source fixtures use TailDrained before cache priming, with normal namespace locks. |
+| `object_transaction_fencing_persists_epoch_on_multipart_commit` | Multipart completion already always drains rename before inspecting all per-disk transaction UUIDs. |
+| `decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page`, `dispatch_completion_cas_is_bounded_and_reaches_the_tail` | Durable receipt, journal, and manifest writers choose TailDrained; the pagination fixture also drains deliberate receipt replacement writes. |
+
+Select these checks with `cargo nextest list -p rustfs-ecstore --features test-util -E 'test(tail_drained_put) | test(early_ack_tail_drain) | test(no_lock_put_waits_for_rename_tail) | test(object_transaction_fencing_persists_epoch_on_multipart_commit) | test(transition_and_restore_reclaim) | test(decommission_durable_ilm_receipt_pagination) | test(dispatch_completion_cas)'`, then run the same expression under the default and CI profiles without retries. Remaining crash, reopen, rollback, and lock-loss schedules use the existing domain tests; this completion fixture is not a replacement for those checks.
+
 ### Coverage gate
 
 `full` and `destructive` run `cargo llvm-cov -p rustfs-ecstore --lib` and fail when line coverage of the gate scope is below `--unit-coverage-min`. The default minimum and the 100% target for EC read, write, decode, heal, metadata-quorum, and rollback paths are the `UNIT_COVERAGE_*` constants at the top of the runner. `cargo-llvm-cov` must be installed unless `--skip-coverage` is passed explicitly. The default scope `ec-critical` is:

rustfs/src/app/bucket_list_through.rs

@@ -252,8 +252,16 @@ pub(crate) async fn merged_list_objects_v2(
             .filter(|entry| merger.accepts(&entry.key().name))
             .collect();
         let keys: Vec<ListEntryKey> = kept.iter().map(SideEntry::key).collect();
+        if let Err(error) = merger.push_page(fetch.side, keys, is_truncated, next_token) {
+            match fetch.side {
+                MergeSide::Source => {
+                    degrade_or_fail(&mut merger, &mut degraded, policy.source_error, "invalid_pagination")?;
+                    continue;
+                }
+                MergeSide::Local => return Err(S3Error::with_message(S3ErrorCode::InternalError, error.to_string())),
+            }
+        }
         buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.into_iter().map(Some));
-        merger.push_page(fetch.side, keys, is_truncated, next_token);
     }
 
     let outcome = merger.finish();
@@ -340,10 +348,12 @@ async fn fetch_source_page(
             continuation_token: token,
             max_keys: params.max_keys,
         },
-        // Everything under `filter.prefix` rolls into one common prefix, so a
-        // single bounded listing settles whether it exists.
+        // Everything under `filter.prefix` rolls into one common prefix. An
+        // empty truncated probe must still follow its cursor before declaring
+        // that prefix absent.
         SourceListPlan::Folded { probe_prefix, .. } => SourceListRequest {
             prefix: Some(probe_prefix.as_str()),
+            continuation_token: token,
             max_keys: 1,
             ..Default::default()
         },
@@ -368,8 +378,8 @@ async fn fetch_source_page(
                 } else {
                     Vec::new()
                 },
-                false,
-                None,
+                !exists && page.is_truncated,
+                if exists { None } else { page.next_continuation_token },
             ))
         }
         _ => {
@@ -417,6 +427,17 @@ async fn local_delete_markers(store: &Arc<ECStore>, bucket: &str, keys: &[String
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::app::bucket_usecase::DefaultBucketUsecase;
+    use crate::app::gating_test_env::{run_large_stack_test, shared_gating_ecstore};
+    use crate::app::storage_api::bucket_usecase::bucket::on_demand_migration::{
+        FilterConfig, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, SourceCredentials, TlsConfig,
+    };
+    use crate::app::storage_api::bucket_usecase::s3::{ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response};
+    use crate::app::storage_api::test::StoragePutObjReader;
+    use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+    use crate::app::storage_api::test::contract::object::ObjectIO as _;
+    use std::time::Duration;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
     fn token(local: Option<&str>, local_done: bool) -> ListThroughToken {
         ListThroughToken {
@@ -526,4 +547,332 @@ mod tests {
         assert!(degraded);
         assert_eq!(merger.next_fetch().map(|fetch| fetch.side), Some(MergeSide::Local));
     }
+
+    /// Serves exactly the scripted S3 pages and joins every connection before
+    /// returning. A source retry or unexpected operation fails the test.
+    async fn scripted_list_source(pages: Vec<String>) -> (String, tokio_util::task::AbortOnDropHandle<Vec<String>>) {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind listing source");
+        let address = listener.local_addr().expect("listing source address");
+        let server = tokio::spawn(async move {
+            let mut requests = Vec::new();
+            for body in pages {
+                let (mut stream, _) = listener.accept().await.expect("accept source listing");
+                let mut request = Vec::new();
+                let mut chunk = [0; 4096];
+                while !request.windows(4).any(|window| window == b"\r\n\r\n") {
+                    let count = stream.read(&mut chunk).await.expect("read signed listing request");
+                    assert!(count > 0, "source request must include complete headers");
+                    request.extend_from_slice(&chunk[..count]);
+                    assert!(request.len() <= 32 * 1024, "listing request headers must be bounded");
+                }
+                let first_line = String::from_utf8_lossy(&request)
+                    .lines()
+                    .next()
+                    .expect("request line")
+                    .to_string();
+                // The SDK joins the bucket endpoint with the LIST operation's `/` path.
+                assert!(
+                    first_line.starts_with("GET /source-bucket/?"),
+                    "expected a path-style bucket-root LIST request, got {first_line:?}"
+                );
+                assert!(first_line.contains("list-type=2"), "expected a ListObjectsV2 query, got {first_line:?}");
+                requests.push(first_line);
+                let response = format!(
+                    "HTTP/1.1 200 OK\r\ncontent-type: application/xml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
+                    body.len()
+                );
+                stream.write_all(response.as_bytes()).await.expect("write source page");
+                stream.shutdown().await.expect("finish source response");
+            }
+            requests
+        });
+        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server))
+    }
+
+    fn source_xml(next: Option<&str>, truncated: bool, key: Option<&str>) -> String {
+        let next = next
+            .map(|token| format!("<NextContinuationToken>{token}</NextContinuationToken>"))
+            .unwrap_or_default();
+        let contents = key
+            .map(|key| format!("<Contents><Key>{key}</Key><Size>1</Size></Contents>"))
+            .unwrap_or_default();
+        format!(
+            "<ListBucketResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><IsTruncated>{truncated}</IsTruncated>{next}{contents}</ListBucketResult>"
+        )
+    }
+
+    struct ListThroughTestState {
+        bucket: String,
+        module_enabled: bool,
+    }
+
+    impl Drop for ListThroughTestState {
+        fn drop(&mut self) {
+            let sys = OnDemandMigrationSys::get();
+            sys.remove(&self.bucket);
+            sys.set_module_enabled(self.module_enabled);
+        }
+    }
+
+    async fn source_policy_request(
+        pages: Vec<String>,
+        policy: SourceErrorPolicy,
+        resume_source: Option<&str>,
+        filter_prefix: Option<&str>,
+    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
+        let store = shared_gating_ecstore().await;
+        crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
+        let bucket = format!("odm-list-{}", uuid::Uuid::new_v4().simple());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create list-through bucket");
+        store
+            .put_object(
+                &bucket,
+                "z-local",
+                &mut StoragePutObjReader::from_vec(vec![1]),
+                &StorageObjectOptions::default(),
+            )
+            .await
+            .expect("seed real local listing");
+        let (endpoint, server) = scripted_list_source(pages).await;
+        let sys = OnDemandMigrationSys::get();
+        let _state_guard = ListThroughTestState {
+            bucket: bucket.clone(),
+            module_enabled: sys.is_module_enabled(),
+        };
+        sys.set_module_enabled(true);
+        let config = OnDemandMigrationConfig {
+            version: 1,
+            enabled: true,
+            source: SourceConfig {
+                provider: Provider::Minio,
+                endpoint: Some(endpoint),
+                region: "us-east-1".into(),
+                bucket: "source-bucket".into(),
+                path_style: PathStyle::Path,
+                credentials: Some(SourceCredentials {
+                    access_key: "test-access".into(),
+                    secret_key: "test-secret".into(),
+                    session_token: None,
+                }),
+                tls: TlsConfig::default(),
+            },
+            filter: FilterConfig {
+                prefix: filter_prefix.map(str::to_string),
+                ..Default::default()
+            },
+            policy: PolicyConfig {
+                list_through: true,
+                source_error: policy,
+                ..Default::default()
+            },
+        };
+        sys.apply(&bucket, Some(&config)).await;
+        assert!(
+            sys.state(&bucket).expect("ODM state installed").client().is_ok(),
+            "fake source client must build"
+        );
+        let continuation_token = resume_source.map(|source| {
+            let token = ListThroughToken {
+                t: "odm-list".into(),
+                v: 1,
+                local: None,
+                local_done: false,
+                source: Some(source.into()),
+                source_done: false,
+                last_key: None,
+            };
+            base64_simd::STANDARD.encode_to_string(token.encode().as_bytes())
+        });
+        let input = ListObjectsV2Input {
+            bucket,
+            max_keys: Some(2),
+            continuation_token,
+            delimiter: filter_prefix.map(|_| "/".to_string()),
+            encoding_type: None,
+            expected_bucket_owner: None,
+            fetch_owner: None,
+            optional_object_attributes: None,
+            prefix: None,
+            request_payer: None,
+            start_after: None,
+        };
+        let request = S3Request {
+            input,
+            method: http::Method::GET,
+            uri: http::Uri::from_static("/?list-type=2"),
+            headers: HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+        let result = tokio::time::timeout(
+            Duration::from_secs(10),
+            DefaultBucketUsecase::from_global().execute_list_objects_v2(request),
+        )
+        .await
+        .expect("listing must complete within its bounded source budget");
+        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+            .await
+            .expect("source connections must finish")
+            .expect("source server must not panic");
+        (result, requests)
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_invalid_source_pagination_obeys_policy_on_the_handler_path() {
+        run_large_stack_test("list-through-source-policy", || async {
+            temp_env::async_with_vars(
+                [
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                        for next in [None, Some(""), Some("stuck")] {
+                            for key in [None, Some("a-source")] {
+                                let (result, requests) =
+                                    source_policy_request(vec![source_xml(next, true, key)], policy, Some("stuck"), None).await;
+                                assert_eq!(requests.len(), 1, "a malformed source page must not be retried");
+                                assert!(requests[0].contains("continuation-token=stuck"));
+                                assert_source_policy_result(result, policy);
+                            }
+                        }
+                        let (result, requests) = source_policy_request(
+                            vec![
+                                source_xml(Some("stuck"), true, Some("a-source")),
+                                source_xml(Some("stuck"), true, None),
+                            ],
+                            policy,
+                            None,
+                            None,
+                        )
+                        .await;
+                        assert_eq!(requests.len(), 2, "the failure must occur during a real refill");
+                        assert!(!requests[0].contains("continuation-token="));
+                        assert!(requests[1].contains("continuation-token=stuck"));
+                        assert_source_policy_result(result, policy);
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_empty_advancing_source_pages_reach_eof_on_the_handler_path() {
+        run_large_stack_test("list-through-empty-source-pages", || async {
+            temp_env::async_with_vars(
+                [
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    for filter_prefix in [None, Some("photos/2024/")] {
+                        let source_key = if filter_prefix.is_some() {
+                            "photos/2024/a-source"
+                        } else {
+                            "a-source"
+                        };
+                        let (result, requests) = source_policy_request(
+                            vec![
+                                source_xml(Some("opaque-next"), true, None),
+                                source_xml(None, false, Some(source_key)),
+                            ],
+                            SourceErrorPolicy::Propagate,
+                            None,
+                            filter_prefix,
+                        )
+                        .await;
+                        assert_eq!(requests.len(), 2, "an empty truncated source page must reach its successor");
+                        assert!(requests[1].contains("continuation-token=opaque-next"));
+                        let response = result.expect("empty progressing source page is valid");
+                        assert!(!response.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                        let output = response.output;
+                        let objects: Vec<_> = output
+                            .contents
+                            .unwrap_or_default()
+                            .into_iter()
+                            .map(|object| object.key.expect("listed object key"))
+                            .collect();
+                        if filter_prefix.is_some() {
+                            assert_eq!(objects, vec!["z-local"]);
+                            assert_eq!(
+                                output
+                                    .common_prefixes
+                                    .unwrap_or_default()
+                                    .into_iter()
+                                    .map(|prefix| prefix.prefix.expect("rolled-up prefix"))
+                                    .collect::<Vec<_>>(),
+                                vec!["photos/"]
+                            );
+                        } else {
+                            assert_eq!(objects, vec!["a-source", "z-local"]);
+                            assert!(output.common_prefixes.unwrap_or_default().is_empty());
+                        }
+                        assert_eq!(output.key_count, Some(2));
+                        assert_eq!(output.is_truncated, Some(false));
+                        assert!(output.next_continuation_token.is_none());
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    fn assert_source_policy_result(result: S3Result<S3Response<ListObjectsV2Output>>, policy: SourceErrorPolicy) {
+        match policy {
+            SourceErrorPolicy::Propagate => {
+                let error = result.expect_err("propagate must expose malformed pagination");
+                assert_eq!(error.status_code(), Some(http::StatusCode::FAILED_DEPENDENCY));
+                assert_eq!(error.code(), &S3ErrorCode::Custom("SourceUnavailable".into()));
+                assert_eq!(error.message(), Some("invalid_pagination"));
+            }
+            SourceErrorPolicy::NotFound => {
+                let response = result.expect("not_found must preserve the local listing");
+                assert_eq!(
+                    response
+                        .headers
+                        .get("x-rustfs-on-demand-migration-list")
+                        .expect("local_only header"),
+                    "local_only"
+                );
+                let output = response.output;
+                assert_eq!(
+                    output
+                        .contents
+                        .unwrap_or_default()
+                        .into_iter()
+                        .map(|object| object.key.expect("local key"))
+                        .collect::<Vec<_>>(),
+                    vec!["z-local"]
+                );
+                assert_eq!(output.is_truncated, Some(false));
+                assert_eq!(output.key_count, Some(1));
+                assert!(output.next_continuation_token.is_none());
+            }
+        }
+    }
 }

rustfs/src/app/storage_api.rs

@@ -29,11 +29,13 @@ pub(crate) fn EndpointServerPools(
 pub(crate) mod s3 {
     #[cfg(test)]
     pub(crate) use s3s::dto::{
-        BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ReplicationConfiguration,
-        ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, ServerSideEncryptionByDefault,
-        ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration,
+        BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ListObjectsV2Input,
+        ListObjectsV2Output, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus,
+        ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration,
     };
     pub(crate) use s3s::{S3Error, S3ErrorCode, S3Result};
+    #[cfg(test)]
+    pub(crate) use s3s::{S3Request, S3Response};
 }
 
 pub(crate) mod admin {

scripts/check_test_wiring.py

@@ -764,8 +764,53 @@ def check_profile_listing(root: Path, profile: str, listing: Path) -> list[str]:
     return []
 
 
+def core_requirements(root: Path) -> dict:
+    data = json.loads((root / ".config/ecstore-required-tests.json").read_text())
+    if not data["tests"] or not data["fixtures"]:
+        raise ValueError("core test and fixture requirements must not be empty")
+    identities = [(test["suite"], test["name"]) for test in data["tests"]]
+    if len(set(identities)) != len(identities):
+        raise ValueError("duplicate core test requirement")
+    return data
+
+
+def check_core_fixtures(root: Path) -> list[str]:
+    try:
+        fixtures = core_requirements(root)["fixtures"]
+        errors = []
+        for fixture in fixtures:
+            path = (root / fixture["path"]).resolve()
+            if not path.is_relative_to(root.resolve()):
+                raise ValueError("core fixture path escapes repository")
+            if not path.is_file():
+                errors.append(f"{fixture['path']}: required core fixture missing")
+            elif hashlib.sha256(path.read_bytes()).hexdigest() != fixture["sha256"]:
+                errors.append(f"{fixture['path']}: core fixture sha256 mismatch")
+        return errors
+    except (OSError, KeyError, TypeError, ValueError) as error:
+        return [f"cannot validate core fixtures: {error}"]
+
+
+def check_core_listing(root: Path, listing: Path) -> list[str]:
+    """Check the existing CI run's selection, not a second filtered test run."""
+    try:
+        required = core_requirements(root)["tests"]
+        suites = json.loads(listing.read_text())["rust-suites"]
+        if not isinstance(suites, dict):
+            raise ValueError("rust-suites must be an object")
+        errors = check_core_fixtures(root)
+        for test in required:
+            testcase = suites.get(test["suite"], {}).get("testcases", {}).get(test["name"], {})
+            if testcase.get("ignored") is not False or testcase.get("filter-match", {}).get("status") != "matches":
+                errors.append(f"{test['invariant']}: required test not selected: {test['suite']}::{test['name']}")
+        return errors
+    except (OSError, KeyError, TypeError, ValueError) as error:
+        return [f"cannot read core nextest listing: {error}"]
+
+
 def validate(root: Path) -> list[str]:
     errors: list[str] = []
+    errors.extend(check_core_fixtures(root))
     errors.extend(check_e2e_modules(root))
     errors.extend(check_vault_test_groups(root))
     errors.extend(check_ilm_build_budget(root))
@@ -779,6 +824,32 @@ def validate(root: Path) -> list[str]:
 
 
 class SelfTests(unittest.TestCase):
+    def test_core_gate_rejects_missing_ignored_filtered_and_corrupt_inputs(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / ".config").mkdir()
+            fixture = root / "fixture.hex"
+            fixture.write_text("4142")
+            requirements = {
+                "tests": [{"invariant": "commit", "suite": "store", "name": "commit_test"}],
+                "fixtures": [{"path": "fixture.hex", "sha256": hashlib.sha256(fixture.read_bytes()).hexdigest()}],
+            }
+            (root / ".config/ecstore-required-tests.json").write_text(json.dumps(requirements))
+            listing = root / "listing.json"
+            good = {"ignored": False, "filter-match": {"status": "matches"}}
+            for case, testcase in (("selected", good), ("missing", {}), ("ignored", dict(good, ignored=True)),
+                                   ("filtered", dict(good, **{"filter-match": {"status": "mismatch"}}))):
+                with self.subTest(case=case):
+                    listing.write_text(json.dumps({"rust-suites": {"store": {"testcases": {"commit_test": testcase}}}}))
+                    self.assertEqual(bool(check_core_listing(root, listing)), case != "selected")
+            listing.write_text(json.dumps({"rust-suites": {"store": {"testcases": {"commit_test": good}}}}))
+            fixture.write_text("4143")
+            self.assertIn("sha256 mismatch", check_core_listing(root, listing)[0])
+            fixture.unlink()
+            self.assertIn("fixture missing", check_core_listing(root, listing)[0])
+            listing.write_text("not json")
+            self.assertIn("cannot read", check_core_listing(root, listing)[0])
+
     def test_ilm_lane_keeps_the_measured_cargo_build_budget(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)
@@ -981,6 +1052,7 @@ class SelfTests(unittest.TestCase):
                 mock.patch(__name__ + ".check_e2e_modules", return_value=[]),
                 mock.patch(__name__ + ".check_vault_test_groups", return_value=[]),
                 mock.patch(__name__ + ".check_fuzz_targets", return_value=[]),
+                mock.patch(__name__ + ".check_core_fixtures", return_value=[]),
                 mock.patch(__name__ + ".check_runner_selection", return_value=[]),
                 mock.patch(__name__ + ".check_workflow_readiness", return_value=[]),
                 mock.patch(__name__ + ".check_profile_definitions", return_value=[]),
@@ -1393,6 +1465,11 @@ def main() -> int:
     if sys.argv[1:] == ["--self-test"]:
         suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)
         return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
+    if len(sys.argv) == 3 and sys.argv[1] == "--check-core":
+        errors = check_core_listing(ROOT, Path(sys.argv[2]))
+        for error in errors:
+            print(f"ERROR: {error}", file=sys.stderr)
+        return 1 if errors else 0
     if len(sys.argv) == 4 and sys.argv[1] == "--check-profile":
         errors = check_profile_listing(ROOT, sys.argv[2], Path(sys.argv[3]))
         if errors:
@@ -1410,7 +1487,7 @@ def main() -> int:
         return 0
     if sys.argv[1:]:
         print(
-            "usage: check_test_wiring.py [--self-test | --check-profile PROFILE LISTING | "
+            "usage: check_test_wiring.py [--self-test | --check-core LISTING | --check-profile PROFILE LISTING | "
             "--update-profile PROFILE LISTING PLATFORM]",
             file=sys.stderr,
         )