fix(ecstore): integrate verified rename preflight evidence

test(ecstore): match sealed context fixture map type

.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/api/mod.rs

@@ -479,9 +479,11 @@ pub mod notification {
     #[cfg(any(test, feature = "test-util"))]
     pub use crate::services::notification_sys::rotate_cross_pool_fence_fleet_proof_for_test;
     pub use crate::services::notification_sys::{
-        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
-        acquire_cross_pool_fence_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
-        new_global_notification_sys, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
+        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, LegacyTransitionStateReconcileFleetProofToken, NotificationPeerErr,
+        NotificationSys, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof,
+        acquire_legacy_transition_state_reconcile_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
+        legacy_transition_state_reconcile_fleet_proof_matches, new_global_notification_sys,
+        scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
     };
 }
 

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/bucket/sealed_credentials.rs

@@ -203,7 +203,7 @@ mod tests {
     use parking_lot::Mutex;
     use std::collections::BTreeMap;
 
-    fn encode_context(context: &HashMap<String, String>) -> String {
+    fn encode_context(context: &BTreeMap<String, String>) -> String {
         let ordered = context.iter().collect::<BTreeMap<_, _>>();
         serde_json::to_string(&ordered).expect("context serializes")
     }

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/notification_sys.rs

@@ -62,12 +62,27 @@ const REMOTE_VERSION_STATE_PROOF_TTL: Duration = Duration::from_secs(30);
 const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 2;
 const TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION: u32 = 3;
 const DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4;
+// Keep this synchronized with the version served by node_service. Including
+// the local member in the minimum prevents an older coordinator from
+// self-authorizing a policy implemented only by newer remote peers.
+const LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4;
+/// Version 5 is reserved for a fleet whose every metadata writer preserves
+/// explicit transition version state and destination identity, and implements
+/// conditional per-generation `xl.meta` writes with strong readback. The node
+/// service must not advertise this version until the conditional writer from
+/// rustfs/backlog#684 is available.
+const LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION: u32 = 5;
 type CrossPoolFencePolicyResult = Result<BTreeMap<String, Uuid>>;
 
 fn cross_pool_fence_policy_results(
     peer_epochs: BTreeMap<String, Uuid>,
     minimum_version: u32,
-) -> (CrossPoolFencePolicyResult, CrossPoolFencePolicyResult, CrossPoolFencePolicyResult) {
+) -> (
+    CrossPoolFencePolicyResult,
+    CrossPoolFencePolicyResult,
+    CrossPoolFencePolicyResult,
+    CrossPoolFencePolicyResult,
+) {
     let journal_result = if minimum_version >= TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION {
         Ok(peer_epochs.clone())
     } else {
@@ -78,7 +93,18 @@ fn cross_pool_fence_policy_results(
     } else {
         Err(Error::other("decommission target fence policy capability version is unsupported"))
     };
-    (Ok(peer_epochs), journal_result, decommission_target_fence_result)
+    let legacy_transition_state_reconcile_result =
+        if minimum_version >= LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION {
+            Ok(peer_epochs.clone())
+        } else {
+            Err(Error::other("legacy transition state reconcile policy capability version is unsupported"))
+        };
+    (
+        Ok(peer_epochs),
+        journal_result,
+        decommission_target_fence_result,
+        legacy_transition_state_reconcile_result,
+    )
 }
 
 #[derive(Clone, Debug)]
@@ -252,10 +278,21 @@ pub(crate) struct TierDeleteJournalFleetProofToken {
     _permit: FleetCapabilityProofPermit,
 }
 
+/// Effect-window authority for one legacy transition-state reconciliation.
+///
+/// The token intentionally cannot be cloned. Its permit keeps the admitted
+/// fleet generation alive until the caller finishes the final strong
+/// readback, while revocation makes every later validation fail immediately.
+pub struct LegacyTransitionStateReconcileFleetProofToken {
+    token: FleetCapabilityProofToken,
+    _permit: FleetCapabilityProofPermit,
+}
+
 static REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static CROSS_POOL_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static TIER_DELETE_JOURNAL_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static DECOMMISSION_TARGET_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
+static LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
 
 fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
@@ -274,6 +311,10 @@ fn decommission_target_fence_fleet_proof_slot() -> &'static std::sync::RwLock<Fl
     DECOMMISSION_TARGET_FENCE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
 }
 
+fn legacy_transition_state_reconcile_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
+    LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
+}
+
 fn revoke_fleet_capability_proof_state(state: &mut FleetCapabilityProofState) {
     if let Some(proof) = state.proof.take() {
         proof.generation.revoke();
@@ -444,6 +485,125 @@ pub(crate) fn tier_delete_journal_topology_generation(proof: &TierDeleteJournalF
     stable_tier_delete_journal_topology_generation(&proof.token.topology_fingerprint)
 }
 
+/// Acquire one non-cloneable authority that must span the complete reconcile
+/// effect window, including its final strong readback.
+pub async fn acquire_legacy_transition_state_reconcile_fleet_proof() -> Option<LegacyTransitionStateReconcileFleetProofToken> {
+    let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
+    let proof = {
+        let state = legacy_transition_state_reconcile_fleet_proof_slot()
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, expected_topology, Instant::now())?
+    };
+    let observed_peer_epochs = observe_legacy_transition_state_reconcile_fleet(expected_topology).await?;
+    let state = legacy_transition_state_reconcile_fleet_proof_slot()
+        .read()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
+        &state,
+        &proof,
+        expected_topology,
+        &observed_peer_epochs,
+        Instant::now(),
+    )
+    .then_some(proof)
+}
+
+fn acquire_legacy_transition_state_reconcile_fleet_proof_from(
+    state: &FleetCapabilityProofState,
+    expected_topology: &str,
+    now: Instant,
+) -> Option<LegacyTransitionStateReconcileFleetProofToken> {
+    let token = acquire_fleet_capability_proof_from(state, expected_topology, now)?;
+    let permit = state.proof.as_ref()?.generation.try_acquire()?;
+    Some(LegacyTransitionStateReconcileFleetProofToken { token, _permit: permit })
+}
+
+async fn observe_legacy_transition_state_reconcile_fleet(expected_topology: &str) -> Option<BTreeMap<String, Uuid>> {
+    let notification_sys = get_global_notification_sys()?;
+    let (peer_epochs, minimum_version) = timeout(
+        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
+        notification_sys.probe_cross_pool_fence_fleet(expected_topology),
+    )
+    .await
+    .ok()?
+    .ok()?;
+    let (_, _, _, reconcile_result) = cross_pool_fence_policy_results(peer_epochs, minimum_version);
+    reconcile_result.ok()
+}
+
+/// Revalidate the exact fleet generation captured by a reconcile token with a
+/// fresh synchronous observation. Callers must await this before each
+/// conditional metadata write and after the final strong readback.
+pub async fn legacy_transition_state_reconcile_fleet_proof_matches(
+    proof: &LegacyTransitionStateReconcileFleetProofToken,
+) -> bool {
+    let Some(expected_topology) = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() else {
+        return false;
+    };
+    legacy_transition_state_reconcile_fleet_proof_matches_with_observer(
+        legacy_transition_state_reconcile_fleet_proof_slot(),
+        proof,
+        expected_topology,
+        || observe_legacy_transition_state_reconcile_fleet(expected_topology),
+    )
+    .await
+}
+
+async fn legacy_transition_state_reconcile_fleet_proof_matches_with_observer<F, Fut>(
+    slot: &std::sync::RwLock<FleetCapabilityProofState>,
+    proof: &LegacyTransitionStateReconcileFleetProofToken,
+    expected_topology: &str,
+    observe: F,
+) -> bool
+where
+    F: FnOnce() -> Fut,
+    Fut: Future<Output = Option<BTreeMap<String, Uuid>>>,
+{
+    {
+        let state = slot.read().unwrap_or_else(std::sync::PoisonError::into_inner);
+        if !legacy_transition_state_reconcile_fleet_proof_matches_at(&state, proof, expected_topology, Instant::now()) {
+            return false;
+        }
+    }
+    let Some(observed_peer_epochs) = observe().await else {
+        return false;
+    };
+    let state = slot.read().unwrap_or_else(std::sync::PoisonError::into_inner);
+    legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
+        &state,
+        proof,
+        expected_topology,
+        &observed_peer_epochs,
+        Instant::now(),
+    )
+}
+
+fn legacy_transition_state_reconcile_fleet_proof_matches_at(
+    state: &FleetCapabilityProofState,
+    proof: &LegacyTransitionStateReconcileFleetProofToken,
+    expected_topology: &str,
+    now: Instant,
+) -> bool {
+    proof._permit.generation.is_accepting()
+        && fleet_capability_proof_matches_at(state, &proof.token, expected_topology, now)
+        && state
+            .proof
+            .as_ref()
+            .is_some_and(|current| Arc::ptr_eq(&current.generation, &proof._permit.generation))
+}
+
+fn legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
+    state: &FleetCapabilityProofState,
+    proof: &LegacyTransitionStateReconcileFleetProofToken,
+    expected_topology: &str,
+    observed_peer_epochs: &BTreeMap<String, Uuid>,
+    now: Instant,
+) -> bool {
+    legacy_transition_state_reconcile_fleet_proof_matches_at(state, proof, expected_topology, now)
+        && proof.token.peer_epochs.as_ref() == observed_peer_epochs
+}
+
 #[cfg(all(test, feature = "test-util"))]
 pub(crate) fn tier_delete_journal_fleet_proof_has_inflight_for_test() -> bool {
     let state = tier_delete_journal_fleet_proof_slot()
@@ -766,6 +926,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 cross_pool_fence_fleet_proof_slot(),
                 tier_delete_journal_fleet_proof_slot(),
                 decommission_target_fence_fleet_proof_slot(),
+                legacy_transition_state_reconcile_fleet_proof_slot(),
             ] {
                 mark_fleet_capability_topology_conflict(slot);
             }
@@ -798,11 +959,12 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 .unwrap_or_else(|_| Err(Error::other("cross-pool fence fleet capability probe timed out"))),
                 None => Err(Error::other("cross-pool fence fleet capability notification system is unavailable")),
             };
-            let (fence_result, journal_result, decommission_target_fence_result) = match fence_probe {
+            let (fence_result, journal_result, decommission_target_fence_result, reconcile_result) = match fence_probe {
                 Ok((peer_epochs, minimum_version)) => cross_pool_fence_policy_results(peer_epochs, minimum_version),
                 Err(err) => {
                     let message = err.to_string();
                     (
+                        Err(Error::other(message.clone())),
                         Err(Error::other(message.clone())),
                         Err(Error::other(message.clone())),
                         Err(Error::other(message)),
@@ -818,6 +980,7 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 revoke_fleet_capability_proof(cross_pool_fence_fleet_proof_slot());
                 revoke_fleet_capability_proof(tier_delete_journal_fleet_proof_slot());
                 revoke_fleet_capability_proof(decommission_target_fence_fleet_proof_slot());
+                revoke_fleet_capability_proof(legacy_transition_state_reconcile_fleet_proof_slot());
             } else if let Some(err) = publish_fleet_capability_probe_result(
                 remote_version_state_fleet_proof_slot(),
                 &topology_fingerprint,
@@ -880,6 +1043,24 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                     "notification capability probe"
                 );
             }
+            if !topology_conflict
+                && let Some(err) = publish_fleet_capability_probe_result(
+                    legacy_transition_state_reconcile_fleet_proof_slot(),
+                    &topology_fingerprint,
+                    reconcile_result,
+                    Instant::now(),
+                )
+            {
+                debug!(
+                    event = EVENT_NOTIFICATION_CAPABILITY_PROBE,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                    capability = "legacy_transition_state_reconcile_v1",
+                    state = "failed_closed",
+                    error = %err,
+                    "notification capability probe"
+                );
+            }
             sleep(REMOTE_VERSION_STATE_PROBE_INTERVAL).await;
         }
     });
@@ -959,7 +1140,7 @@ impl NotificationSys {
             client.probe_cross_pool_fence(topology_fingerprint.to_string()).await
         });
         let mut peer_epochs = BTreeMap::new();
-        let mut minimum_version = u32::MAX;
+        let mut minimum_version = LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION;
         for result in join_all(probes).await {
             let (peer, version, epoch) = result?;
             if version < CROSS_POOL_FENCE_SUPPORTED_VERSION {
@@ -968,11 +1149,6 @@ impl NotificationSys {
             minimum_version = minimum_version.min(version);
             insert_remote_version_state_peer(&mut peer_epochs, peer, epoch)?;
         }
-        // A single-node deployment has no remote member to lower the local
-        // policy version advertised by this binary.
-        if minimum_version == u32::MAX {
-            minimum_version = DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION;
-        }
         Ok((peer_epochs, minimum_version))
     }
 }
@@ -3190,20 +3366,36 @@ mod tests {
     #[test]
     fn cross_pool_policy_versions_authorize_only_their_supported_protocols() {
         let peers = BTreeMap::from([("node-b:9000".to_string(), Uuid::new_v4())]);
-        let (generic_v2, journal_v2, decommission_v2) = cross_pool_fence_policy_results(peers.clone(), 2);
+        let (generic_v2, journal_v2, decommission_v2, reconcile_v2) = cross_pool_fence_policy_results(peers.clone(), 2);
         assert!(generic_v2.is_ok(), "v2 remains valid for existing cross-pool fencing");
         assert!(journal_v2.is_err(), "a mixed v2/v3 fleet must fail closed for journal-v6 deletion");
         assert!(decommission_v2.is_err(), "v2 cannot authorize the sticky per-target decommission fence");
+        assert!(reconcile_v2.is_err(), "v2 cannot authorize legacy transition-state reconciliation");
 
-        let (generic_v3, journal_v3, decommission_v3) = cross_pool_fence_policy_results(peers.clone(), 3);
+        let (generic_v3, journal_v3, decommission_v3, reconcile_v3) = cross_pool_fence_policy_results(peers.clone(), 3);
         assert!(generic_v3.is_ok());
         assert!(journal_v3.is_ok(), "an all-v3 fleet may authorize journal-v6 deletion");
         assert!(decommission_v3.is_err(), "v3 members do not understand the per-target decommission fence");
+        assert!(reconcile_v3.is_err());
 
-        let (generic_v4, journal_v4, decommission_v4) = cross_pool_fence_policy_results(peers, 4);
+        let (generic_v4, journal_v4, decommission_v4, reconcile_v4) =
+            cross_pool_fence_policy_results(peers.clone(), LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION);
         assert!(generic_v4.is_ok());
         assert!(journal_v4.is_ok());
         assert!(decommission_v4.is_ok(), "an all-v4 fleet may create sticky per-target reservations");
+        assert!(
+            reconcile_v4.is_err(),
+            "the current local policy lacks the conditional xl.meta writer required by reconcile"
+        );
+
+        let (generic_v5, journal_v5, decommission_v5, reconcile_v5) = cross_pool_fence_policy_results(peers, 5);
+        assert!(generic_v5.is_ok());
+        assert!(journal_v5.is_ok());
+        assert!(decommission_v5.is_ok());
+        assert!(
+            reconcile_v5.is_ok(),
+            "only an all-v5 fleet preserves destination identity and conditional reconcile writes"
+        );
     }
 
     #[test]
@@ -3458,6 +3650,234 @@ mod tests {
         );
     }
 
+    #[test]
+    fn legacy_transition_state_reconcile_admits_only_compatible_single_and_multi_node_fleets() {
+        let now = Instant::now();
+        for peers in [
+            BTreeMap::new(),
+            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4()), ("peer-b".to_string(), Uuid::new_v4())]),
+        ] {
+            let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+            let (_, _, _, result) =
+                cross_pool_fence_policy_results(peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+            assert!(publish_fleet_capability_probe_result(&slot, "topology-a", result, now).is_none());
+
+            let admitted = {
+                let state = slot.read().expect("reconcile proof slot should not poison");
+                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                    .expect("an all-compatible fleet should admit reconciliation")
+            };
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            assert!(legacy_transition_state_reconcile_fleet_proof_matches_at(
+                &state,
+                &admitted,
+                "topology-a",
+                now,
+            ));
+        }
+    }
+
+    #[test]
+    fn legacy_transition_state_reconcile_restart_drains_concurrent_effect_windows() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let original_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let (_, _, _, original_result) =
+            cross_pool_fence_policy_results(original_peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", original_result, now).is_none());
+
+        let (first, second) = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            (
+                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                    .expect("the first reconcile writer should be admitted"),
+                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                    .expect("the second reconcile writer should be admitted"),
+            )
+        };
+
+        let restarted_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let (_, _, _, restarted_result) =
+            cross_pool_fence_policy_results(restarted_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        let blocked =
+            publish_fleet_capability_probe_result(&slot, "topology-a", restarted_result, now + Duration::from_millis(1))
+                .expect("a restarted member must revoke the old generation and wait for both writers");
+        assert!(blocked.to_string().contains("previous generation to drain"));
+        {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            assert!(state.proof.is_none());
+            assert!(state.draining_generation.is_some());
+            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+                &state,
+                &first,
+                "topology-a",
+                now + Duration::from_millis(1),
+            ));
+            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+                &state,
+                &second,
+                "topology-a",
+                now + Duration::from_millis(1),
+            ));
+        }
+
+        drop(first);
+        let (_, _, _, still_blocked_result) =
+            cross_pool_fence_policy_results(restarted_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(
+            publish_fleet_capability_probe_result(&slot, "topology-a", still_blocked_result, now + Duration::from_millis(2),)
+                .is_some(),
+            "one remaining writer must keep the successor generation closed"
+        );
+
+        drop(second);
+        let (_, _, _, admitted_result) =
+            cross_pool_fence_policy_results(restarted_peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(
+            publish_fleet_capability_probe_result(&slot, "topology-a", admitted_result, now + Duration::from_millis(3),)
+                .is_none(),
+            "the restarted generation may publish only after every old writer drains"
+        );
+    }
+
+    #[test]
+    fn legacy_transition_state_reconcile_fresh_observation_closes_the_polling_window() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let original_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let (_, _, _, original_result) =
+            cross_pool_fence_policy_results(original_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", original_result, now).is_none());
+        let admitted = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                .expect("the original fleet should admit reconciliation")
+        };
+
+        let restarted_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let state = slot.read().expect("reconcile proof slot should not poison");
+        assert!(
+            legacy_transition_state_reconcile_fleet_proof_matches_at(&state, &admitted, "topology-a", now),
+            "the periodic cache has not observed the restart yet"
+        );
+        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
+            &state,
+            &admitted,
+            "topology-a",
+            &restarted_peers,
+            now,
+        ));
+
+        let (_, _, _, downgraded) = cross_pool_fence_policy_results(original_peers, 4);
+        assert!(
+            downgraded.is_err(),
+            "a synchronous observation of a downgraded peer must fail before any cached proof can authorize a write"
+        );
+    }
+
+    #[tokio::test]
+    async fn legacy_transition_state_reconcile_invalid_token_skips_fleet_observation() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(peers), now).is_none());
+        let admitted = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                .expect("the original fleet should admit reconciliation")
+        };
+        revoke_fleet_capability_proof(&slot);
+
+        assert!(
+            !legacy_transition_state_reconcile_fleet_proof_matches_with_observer(&slot, &admitted, "topology-a", || async {
+                panic!("an invalid local generation must not trigger a fleet observation");
+            },)
+            .await
+        );
+    }
+
+    #[test]
+    fn legacy_transition_state_reconcile_membership_and_topology_changes_revoke_authority() {
+        let now = Instant::now();
+        for replacement in [
+            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4()), ("peer-b".to_string(), Uuid::new_v4())]),
+            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]),
+        ] {
+            let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+            let original = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+            assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(original), now).is_none());
+            let admitted = {
+                let state = slot.read().expect("reconcile proof slot should not poison");
+                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                    .expect("the original fleet should admit reconciliation")
+            };
+
+            assert!(
+                publish_fleet_capability_probe_result(&slot, "topology-a", Ok(replacement), now + Duration::from_millis(1),)
+                    .is_some(),
+                "membership or process-epoch replacement must wait for the admitted writer"
+            );
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+                &state,
+                &admitted,
+                "topology-a",
+                now + Duration::from_millis(1),
+            ));
+        }
+
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(BTreeMap::new()), now).is_none());
+        let admitted = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                .expect("the original topology should admit reconciliation")
+        };
+        mark_fleet_capability_topology_conflict(&slot);
+        let state = slot.read().expect("reconcile proof slot should not poison");
+        assert!(state.topology_conflict);
+        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+            &state,
+            &admitted,
+            "topology-a",
+            now,
+        ));
+    }
+
+    #[test]
+    fn legacy_transition_state_reconcile_capability_downgrade_fails_closed() {
+        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
+        let now = Instant::now();
+        let peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
+        let (_, _, _, compatible_result) =
+            cross_pool_fence_policy_results(peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
+        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", compatible_result, now).is_none());
+        let admitted = {
+            let state = slot.read().expect("reconcile proof slot should not poison");
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
+                .expect("v5 should admit reconciliation")
+        };
+
+        let (_, _, _, downgraded_result) =
+            cross_pool_fence_policy_results(peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION - 1);
+        let err = publish_fleet_capability_probe_result(&slot, "topology-a", downgraded_result, now + Duration::from_millis(1))
+            .expect("a v4 member must revoke reconcile authority");
+        assert!(err.to_string().contains("reconcile policy capability version is unsupported"));
+        let state = slot.read().expect("reconcile proof slot should not poison");
+        assert!(state.proof.is_none());
+        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
+            &state,
+            &admitted,
+            "topology-a",
+            now + Duration::from_millis(1),
+        ));
+        assert!(
+            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now + Duration::from_millis(1),)
+                .is_none(),
+            "a downgraded fleet must remain inspect-only"
+        );
+    }
+
     #[test]
     fn remote_version_state_fleet_proof_conflict_revokes_atomic_snapshot() {
         let now = Instant::now();
@@ -3539,6 +3959,57 @@ mod tests {
         assert!(err.to_string().contains("incomplete"));
     }
 
+    #[tokio::test]
+    async fn legacy_transition_state_reconcile_probe_rejects_missing_or_unreachable_members() {
+        let missing = NotificationSys {
+            peer_clients: Vec::new(),
+            all_peer_clients: vec![None],
+            peer_topology_hosts: vec!["peer-a".to_string()],
+            peer_admin_caches: Vec::new(),
+            tier_config_reload_workers: Default::default(),
+        };
+        let missing_err = missing
+            .probe_cross_pool_fence_fleet("topology-a")
+            .await
+            .expect_err("a missing member slot must prevent reconcile capability proof");
+        assert!(missing_err.to_string().contains("incomplete"));
+
+        let unreachable = NotificationSys {
+            peer_clients: vec![None],
+            all_peer_clients: vec![None, None],
+            peer_topology_hosts: vec!["peer-a".to_string()],
+            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
+            tier_config_reload_workers: Default::default(),
+        };
+        let unreachable_err = unreachable
+            .probe_cross_pool_fence_fleet("topology-a")
+            .await
+            .expect_err("an unreachable member must prevent reconcile capability proof");
+        assert!(unreachable_err.to_string().contains("unreachable"));
+    }
+
+    #[tokio::test]
+    async fn legacy_transition_state_reconcile_single_node_stays_closed_before_local_cas_support() {
+        let notification_sys = NotificationSys {
+            peer_clients: Vec::new(),
+            all_peer_clients: vec![None],
+            peer_topology_hosts: Vec::new(),
+            peer_admin_caches: Vec::new(),
+            tier_config_reload_workers: Default::default(),
+        };
+        let (peers, minimum_version) = notification_sys
+            .probe_cross_pool_fence_fleet("topology-a")
+            .await
+            .expect("a single-node capability probe should complete");
+        assert!(peers.is_empty());
+        assert_eq!(minimum_version, LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION);
+        let (_, _, _, reconcile_result) = cross_pool_fence_policy_results(peers, minimum_version);
+        assert!(
+            reconcile_result.is_err(),
+            "the current node must not self-authorize reconcile before the conditional writer lands"
+        );
+    }
+
     fn build_props(endpoint: &str) -> ServerProperties {
         ServerProperties {
             endpoint: endpoint.to_string(),

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

crates/scanner/src/scanner_folder.rs

@@ -230,7 +230,7 @@ fn scanner_abandoned_child_list_options() -> ListPathRawOptions {
 }
 
 pub fn data_usage_update_dir_cycles() -> u32 {
-    rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES).max(1)
+    rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES)
 }
 
 pub fn heal_object_select_prob() -> u32 {
@@ -806,7 +806,6 @@ impl FolderScanner {
     fn prune_failed_objects_cache(&mut self) {
         let ttl = self.failed_object_ttl_secs;
         if ttl == 0 {
-            self.new_cache.info.failed_objects.clear();
             return;
         }
 
@@ -964,27 +963,6 @@ impl FolderScanner {
         }
     }
 
-    async fn preserve_failed_child(
-        &mut self,
-        parent: &Option<DataUsageHash>,
-        child_hash: &DataUsageHash,
-        parent_entry: &mut DataUsageEntry,
-        child_entry: &DataUsageEntry,
-    ) {
-        // A failed walk proves neither deletion nor a complete replacement.
-        // Keep the previous subtree and mark this snapshot incomplete even
-        // when the failed-object retry cache is disabled or at capacity.
-        parent_entry.failed_objects = parent_entry.failed_objects.saturating_add(1);
-        if self.old_cache.cache.contains_key(&child_hash.key()) {
-            self.new_cache.delete_recursive(child_hash);
-            self.new_cache.copy_with_children(&self.old_cache, child_hash, parent);
-            parent_entry.add_child(child_hash);
-        } else {
-            self.preserve_partial_child_progress(parent, child_hash, parent_entry, child_entry)
-                .await;
-        }
-    }
-
     fn alert_excessive_folders(&self, folder: &str, total_folders: usize) {
         let threshold = scanner_excess_folders_threshold();
         if u64::try_from(total_folders).unwrap_or(u64::MAX) <= threshold {
@@ -1199,6 +1177,8 @@ impl FolderScanner {
                 return Err(ScannerError::Other("Operation cancelled".to_string()));
             }
 
+            self.prune_failed_objects_cache();
+
             let mut abandoned_children: DataUsageHashMap = HashSet::new();
             if !into.compacted {
                 abandoned_children = self.old_cache.find_children_copy(this_hash.clone());
@@ -1241,9 +1221,7 @@ impl FolderScanner {
             };
             let active_object_lock = self.old_cache.info.object_lock.clone();
 
-            ctx.run_until_cancelled(self.sleeper.sleep_folder())
-                .await
-                .ok_or_else(|| ScannerError::Other("Operation cancelled".to_string()))?;
+            self.sleeper.sleep_folder().await;
 
             let mut existing_folders: Vec<CachedFolder> = Vec::new();
             let mut new_folders: Vec<CachedFolder> = Vec::new();
@@ -1470,7 +1448,7 @@ impl FolderScanner {
 
                 let heal_enabled = this_hash.mod_alt(
                     self.old_cache.info.next_cycle as u32 / folder.object_heal_prob_div,
-                    (self.heal_object_select / folder.object_heal_prob_div).max(1),
+                    self.heal_object_select / folder.object_heal_prob_div,
                 ) && self.should_heal().await;
 
                 let mut item = ScannerItem {
@@ -1487,10 +1465,12 @@ impl FolderScanner {
                     file_type: entry_type,
                 };
 
-                // Count unresolved objects in each snapshot without extending
-                // the retry TTL or emitting another failure event.
+                // If this path is already known as failed, just skip it.
+                // We intentionally do NOT call `record_failed` or bump `failed_objects` here,
+                // because the failure was recorded when the original error occurred
+                // (e.g. in the get_size error branch below). This branch only accounts
+                // for subsequent skips of already-failed paths.
                 if self.should_skip_failed(&item.path) {
-                    into.failed_objects = into.failed_objects.saturating_add(1);
                     continue;
                 }
 
@@ -1505,7 +1485,7 @@ impl FolderScanner {
 
                         if failure_action != GetSizeFailureAction::Skip {
                             // Track failed objects to prevent infinite retry loops
-                            into.failed_objects = into.failed_objects.saturating_add(1);
+                            into.failed_objects += 1;
                             self.record_failed(&item.path);
 
                             if should_log_failed_object(into.failed_objects) {
@@ -1584,15 +1564,12 @@ impl FolderScanner {
                             }
                         }
 
-                        ctx.run_until_cancelled(timer.sleep())
-                            .await
-                            .ok_or_else(|| ScannerError::Other("Operation cancelled".to_string()))?;
+                        timer.sleep().await;
                         continue;
                     }
                 };
 
                 found_object_metadata = true;
-                self.new_cache.info.failed_objects.remove(&item.path);
 
                 item.transform_meta_dir();
 
@@ -1604,9 +1581,7 @@ impl FolderScanner {
                 object_count += 1;
                 self.budget.record_object_scanned();
 
-                ctx.run_until_cancelled(timer.sleep())
-                    .await
-                    .ok_or_else(|| ScannerError::Other("Operation cancelled".to_string()))?;
+                timer.sleep().await;
 
                 if ctx.is_cancelled() {
                     return Err(ScannerError::Other("Operation cancelled".to_string()));
@@ -1647,9 +1622,9 @@ impl FolderScanner {
             if self.is_erasure_mode && found_erasure_data_directory && !found_object_metadata {
                 found_object_metadata = true;
                 let metadata_path = path_join_buf(&[&dir_path, STORAGE_FORMAT_FILE]);
-                into.failed_objects = into.failed_objects.saturating_add(1);
 
                 if !self.should_skip_failed(&metadata_path) {
+                    into.failed_objects = into.failed_objects.saturating_add(1);
                     self.record_failed(&metadata_path);
 
                     let failed_cache_entries = self.new_cache.info.failed_objects.len();
@@ -1860,7 +1835,6 @@ impl FolderScanner {
                             error = %e,
                             "Scanner child folder scan failed"
                         );
-                        self.preserve_failed_child(&folder_item.parent, &h, into, &dst).await;
                         continue;
                     }
                     tokio::task::yield_now().await;
@@ -2256,7 +2230,6 @@ impl FolderScanner {
                                 error = %e,
                                 "Scanner heal child folder scan failed"
                             );
-                            self.preserve_failed_child(&folder_item.parent, &h, into, &dst).await;
                             continue;
                         }
                         tokio::task::yield_now().await;
@@ -2423,9 +2396,6 @@ pub async fn scan_data_folder(
     };
 
     let now = FolderScanner::now_secs();
-    // Prune once per bucket walk, not once per directory. Per-path TTL checks
-    // still allow retries during long scans, and insertions enforce the cap.
-    scanner.prune_failed_objects_cache();
     prune_size_reconciliation(&mut scanner.new_cache.info, now);
     prune_size_reconciliation(&mut scanner.update_cache.info, now);
 
@@ -2452,9 +2422,7 @@ pub async fn scan_data_folder(
             new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
             new_cache.info.last_update = Some(SystemTime::now());
             new_cache.info.next_cycle = cache.info.next_cycle;
-            let unresolved_objects = new_cache
-                .size_recursive(&cache.info.name)
-                .is_none_or(|root| root.failed_objects > 0)
+            let unresolved_objects = root.failed_objects > 0
                 || !new_cache.info.failed_objects.is_empty()
                 || !new_cache.info.size_reconciliation.is_empty();
             new_cache.info.snapshot_complete = !unresolved_objects;

crates/scanner/src/scanner_folder/item_actions.rs

@@ -1366,104 +1366,6 @@ mod tests {
         assert_eq!(item.object_path(), "object");
     }
 
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn scanner_blocked_expiry_preserves_usage_replication_and_integrity_work() {
-        use s3s::dto::{LifecycleExpiration, LifecycleRule};
-
-        let lifecycle = Arc::new(BucketLifecycleConfiguration {
-            rules: vec![LifecycleRule {
-                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
-                expiration: Some(LifecycleExpiration {
-                    days: Some(1),
-                    ..Default::default()
-                }),
-                abort_incomplete_multipart_upload: None,
-                del_marker_expiration: None,
-                filter: None,
-                id: None,
-                noncurrent_version_expiration: None,
-                noncurrent_version_transitions: None,
-                prefix: None,
-                transitions: None,
-            }],
-            ..Default::default()
-        });
-        let attempts = |report: &rustfs_scanner_metrics::metrics::ScannerMetricsReport, source: ScannerWorkSource| {
-            report
-                .source_work
-                .iter()
-                .filter(|work| work.source == source.as_str())
-                .map(|work| work.queued + work.skipped + work.missed)
-                .sum::<u64>()
-        };
-        for with_lifecycle in [false, true] {
-            for scan_mode in [HealScanMode::Normal, HealScanMode::Deep] {
-                for guard in ["pending", "failed", "legal_hold"] {
-                    let mut metadata = HashMap::new();
-                    let replication_status = match guard {
-                        "pending" => ReplicationStatusType::Pending,
-                        "failed" => ReplicationStatusType::Failed,
-                        _ => {
-                            metadata.insert("x-amz-object-lock-legal-hold".to_string(), "ON".to_string());
-                            ReplicationStatusType::Completed
-                        }
-                    };
-                    let object = ObjectInfo {
-                        bucket: "bucket".to_string(),
-                        name: "object".to_string(),
-                        version_id: Some(uuid::Uuid::new_v4()),
-                        num_versions: 1,
-                        is_latest: true,
-                        mod_time: Some(OffsetDateTime::now_utc() - time::Duration::days(90)),
-                        size: 4096,
-                        actual_size: 4096,
-                        replication_status,
-                        user_defined: Arc::new(metadata),
-                        ..Default::default()
-                    };
-                    let events = Evaluator::new(lifecycle.clone())
-                        .eval(&[crate::ecstore_object_opts_from_object_info(&object)])
-                        .await
-                        .expect("evaluate expiry guard");
-                    assert_eq!(events[0].action, IlmAction::NoneAction, "expiry must be blocked by {guard}");
-
-                    let mut item = scanner_item_with_prefix("");
-                    item.object_name = "object".to_string();
-                    item.lifecycle = with_lifecycle.then(|| lifecycle.clone());
-                    item.replication = Some(Arc::new(ReplicationConfig::new(None, None)));
-                    item.heal_enabled = true;
-                    item.heal_bitrot = scan_mode == HealScanMode::Deep;
-                    let before = global_metrics().report().await;
-                    let mut summary = SizeSummary::default();
-                    item.apply_actions(vec![object], None, VersioningConfiguration::default(), &[], &mut summary)
-                        .await;
-                    let after = global_metrics().report().await;
-                    assert_eq!(summary.total_size, 4096, "blocked expiry must retain bytes for {guard}");
-                    assert_eq!(summary.versions, 1);
-                    assert_eq!(summary.delete_markers, 0);
-                    assert!(summary.size_reconciliation.is_empty());
-                    assert_eq!(
-                        attempts(&after, scanner_heal_source(scan_mode)) - attempts(&before, scanner_heal_source(scan_mode)),
-                        1,
-                        "integrity work must continue with lifecycle={with_lifecycle}, guard={guard}"
-                    );
-                    assert_eq!(
-                        attempts(&after, ScannerWorkSource::BucketReplication)
-                            - attempts(&before, ScannerWorkSource::BucketReplication),
-                        1,
-                        "replication inspection must continue with lifecycle={with_lifecycle}, guard={guard}"
-                    );
-                    assert_eq!(
-                        attempts(&after, ScannerWorkSource::Lifecycle) - attempts(&before, ScannerWorkSource::Lifecycle),
-                        0,
-                        "blocked expiry must not enqueue destructive lifecycle work"
-                    );
-                }
-            }
-        }
-    }
-
     #[test]
     fn unknown_tier_never_triggers_transition() {
         let object = ObjectInfo {

crates/scanner/src/scanner_folder/tests.rs

@@ -1883,323 +1883,6 @@ async fn test_scan_folder_skips_unreadable_child_directory() {
     assert!(result.is_ok(), "expected unreadable child directory to be skipped");
 }
 
-#[tokio::test]
-#[serial]
-async fn scanner_failed_child_retains_usage_and_scans_healthy_sibling() {
-    for with_prior in [false, true] {
-        let (mut scanner, temp_dir) = build_test_scanner().await;
-        let _guard = TestGuard::new(0, 0, &mut scanner, temp_dir.clone());
-        let bad_dir = temp_dir.join("bucket/bad");
-        tokio::fs::create_dir_all(&bad_dir).await.expect("create failing directory");
-        write_test_object_metadata_bytes(
-            &temp_dir,
-            "bucket",
-            "good",
-            &metadata_for_object_version("bucket", "good", Some(Uuid::new_v4())),
-        )
-        .await;
-        scanner.old_cache.info.name = "bucket".to_string();
-        scanner.new_cache.info.name = "bucket".to_string();
-        scanner.update_cache.info.name = "bucket".to_string();
-        let root_hash = hash_path("bucket");
-        let bad_hash = hash_path("bucket/bad");
-        let mut prior = DataUsageEntry {
-            size: 4096,
-            objects: 2,
-            versions: 3,
-            delete_markers: 1,
-            ..Default::default()
-        };
-        prior.replication_stats = Some(rustfs_data_usage::ReplicationAllStats {
-            replica_size: 4096,
-            replica_count: 2,
-            ..Default::default()
-        });
-        prior.add_tier_sizes(&HashMap::from([(
-            "WARM".to_string(),
-            TierStats {
-                total_size: 4096,
-                num_versions: 3,
-                num_objects: 2,
-            },
-        )]));
-        scanner
-            .old_cache
-            .replace_hashed(&root_hash, &None, &DataUsageEntry::default());
-        if with_prior {
-            scanner.old_cache.replace_hashed(&bad_hash, &Some(root_hash.clone()), &prior);
-        } else {
-            prior = DataUsageEntry::default();
-        }
-        scanner.update_current_path = Arc::new(move |path| {
-            if path == "bucket/bad" {
-                // Replace the directory after enumeration but before descent. This
-                // injects a real read_dir error even when tests run as root.
-                std::fs::remove_dir(&bad_dir).expect("remove enumerated directory");
-                std::fs::write(&bad_dir, b"not a directory").expect("replace enumerated directory");
-            }
-            Box::pin(async {})
-        });
-        let mut root = DataUsageEntry::default();
-        scanner
-            .scan_folder(
-                CancellationToken::new(),
-                CachedFolder {
-                    name: "bucket".to_string(),
-                    parent: None,
-                    object_heal_prob_div: 1,
-                },
-                &mut root,
-            )
-            .await
-            .expect("one failed directory must not stop healthy siblings");
-        let total = scanner.new_cache.size_recursive(&root_hash.key()).expect("root usage");
-        assert_eq!(total.size, prior.size + 1, "unreadable child must retain its previous bytes");
-        assert_eq!(total.objects, prior.objects + 1, "healthy sibling must still be counted");
-        assert_eq!(total.versions, prior.versions + 1);
-        assert_eq!(total.delete_markers, prior.delete_markers);
-        assert_eq!(total.failed_objects, 1, "walk error must keep the snapshot incomplete");
-        assert_eq!(
-            serde_json::to_value(&total.replication_stats).expect("serialize replication usage"),
-            serde_json::to_value(&prior.replication_stats).expect("serialize prior replication usage")
-        );
-        assert_eq!(
-            serde_json::to_value(&total.all_tier_stats).expect("serialize tier usage"),
-            serde_json::to_value(&prior.all_tier_stats).expect("serialize prior tier usage")
-        );
-    }
-}
-
-#[tokio::test]
-#[serial]
-async fn scanner_nested_metadata_failure_without_retry_cache_is_partial_then_recovers() {
-    let (scanner, temp_dir) = build_test_scanner().await;
-    let _guard = TestGuard {
-        temp_dir: Some(temp_dir.clone()),
-    };
-    write_test_object_metadata_bytes(&temp_dir, "bucket", "prefix/bad", b"").await;
-    write_test_object_metadata_bytes(
-        &temp_dir,
-        "bucket",
-        "prefix/good",
-        &metadata_for_object_version("bucket", "prefix/good", Some(Uuid::new_v4())),
-    )
-    .await;
-    temp_env::async_with_vars([(ENV_FAILED_OBJECT_TTL_SECS, Some("0"))], async {
-        for inherited_failure in [false, true] {
-            write_test_object_metadata_bytes(&temp_dir, "bucket", "prefix/bad", b"").await;
-            let mut cache = DataUsageCache {
-                info: DataUsageCacheInfo {
-                    name: "bucket".to_string(),
-                    next_cycle: u64::from(
-                        (0..DATA_USAGE_UPDATE_DIR_CYCLES)
-                            .find(|cycle| !hash_path("bucket/prefix").mod_(*cycle, DATA_USAGE_UPDATE_DIR_CYCLES))
-                            .expect("cycle outside the prefix compaction sample"),
-                    ),
-                    ..Default::default()
-                },
-                ..Default::default()
-            };
-            if inherited_failure {
-                cache
-                    .info
-                    .failed_objects
-                    .insert("removed-object/xl.meta".to_string(), FolderScanner::now_secs());
-            }
-            let budget = ScannerCycleBudget::new(&CancellationToken::new(), Default::default());
-            let result = scan_data_folder(
-                budget.token(),
-                budget,
-                vec![scanner.local_disk.clone()],
-                scanner.local_disk.clone(),
-                cache,
-                None,
-                HealScanMode::Normal,
-                DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
-            )
-            .await;
-            let mut partial = match result {
-                Err(ScannerError::PartialCache(cache)) => *cache,
-                other => panic!("nested failure must never publish a complete snapshot: {other:?}"),
-            };
-            assert!(!partial.info.snapshot_complete);
-            assert!(partial.info.failed_objects.is_empty(), "TTL zero disables only the retry cache");
-            let total = partial.size_recursive("bucket").expect("partial root");
-            assert_eq!(total.objects, 1);
-            assert_eq!(total.failed_objects, 1);
-
-            // Reusing the partial compacted subtree must remain partial even
-            // without a retry ledger. Recovery happens on its next selected cycle.
-            let budget = ScannerCycleBudget::new(&CancellationToken::new(), Default::default());
-            let reused = scan_data_folder(
-                budget.token(),
-                budget,
-                vec![scanner.local_disk.clone()],
-                scanner.local_disk.clone(),
-                partial.clone(),
-                None,
-                HealScanMode::Normal,
-                DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
-            )
-            .await;
-            assert!(matches!(reused, Err(ScannerError::PartialCache(_))));
-            partial.info.next_cycle = u64::from(
-                (0..DATA_USAGE_UPDATE_DIR_CYCLES)
-                    .find(|cycle| hash_path("bucket/prefix").mod_(*cycle, DATA_USAGE_UPDATE_DIR_CYCLES))
-                    .expect("next selected directory cycle"),
-            );
-            write_test_object_metadata_bytes(
-                &temp_dir,
-                "bucket",
-                "prefix/bad",
-                &metadata_for_object_version("bucket", "prefix/bad", Some(Uuid::new_v4())),
-            )
-            .await;
-            let budget = ScannerCycleBudget::new(&CancellationToken::new(), Default::default());
-            let recovered = scan_data_folder(
-                budget.token(),
-                budget,
-                vec![scanner.local_disk.clone()],
-                scanner.local_disk.clone(),
-                partial,
-                None,
-                HealScanMode::Normal,
-                DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
-            )
-            .await
-            .expect("repaired subtree must converge on its next selected cycle");
-            assert!(recovered.info.snapshot_complete);
-            let total = recovered.size_recursive("bucket").expect("recovered root");
-            assert_eq!(total.objects, 2);
-            assert_eq!(total.size, 2);
-            assert_eq!(total.versions, 2);
-            assert_eq!(total.failed_objects, 0);
-        }
-    })
-    .await;
-}
-
-#[tokio::test]
-#[serial]
-async fn scanner_compacted_directory_keeps_aggressive_heal_and_bitrot_sampling() {
-    for scan_mode in [HealScanMode::Normal, HealScanMode::Deep] {
-        for select_prob in [0, 1, 8, 16] {
-            let (mut scanner, temp_dir) = build_test_scanner().await;
-            let _guard = TestGuard::new(0, 0, &mut scanner, temp_dir.clone());
-            write_test_object_metadata_bytes(
-                &temp_dir,
-                "bucket",
-                "object",
-                &metadata_for_object_version("bucket", "object", Some(Uuid::new_v4())),
-            )
-            .await;
-            scanner.old_cache.info.name = "bucket".to_string();
-            scanner.new_cache.info.name = "bucket".to_string();
-            scanner.update_cache.info.name = "bucket".to_string();
-            scanner.is_erasure_mode = true;
-            scanner.heal_object_select = select_prob;
-            scanner.scan_mode = scan_mode;
-            let root_hash = hash_path("bucket");
-            let object_hash = hash_path("bucket/object");
-            scanner.old_cache.info.next_cycle = u64::from(
-                (0..DATA_USAGE_UPDATE_DIR_CYCLES)
-                    .find(|cycle| object_hash.mod_(*cycle, DATA_USAGE_UPDATE_DIR_CYCLES))
-                    .expect("selected directory cycle"),
-            );
-            scanner
-                .old_cache
-                .replace_hashed(&root_hash, &None, &DataUsageEntry::default());
-            scanner.old_cache.replace_hashed(
-                &object_hash,
-                &Some(root_hash),
-                &DataUsageEntry {
-                    compacted: true,
-                    objects: 1,
-                    versions: 1,
-                    ..Default::default()
-                },
-            );
-            let attempts = |report: rustfs_scanner_metrics::metrics::ScannerMetricsReport| {
-                report
-                    .source_work
-                    .iter()
-                    .filter(|work| work.source == scanner_heal_source(scan_mode).as_str())
-                    .map(|work| work.queued + work.skipped + work.missed)
-                    .sum::<u64>()
-            };
-            temp_env::async_with_vars([(ENV_SCANNER_DEEP_VERIFY_COOLDOWN_SECS, Some("0"))], async {
-                let before = attempts(global_metrics().report().await);
-                let mut root = DataUsageEntry::default();
-                scanner
-                    .scan_folder(
-                        CancellationToken::new(),
-                        CachedFolder {
-                            name: "bucket".to_string(),
-                            parent: None,
-                            object_heal_prob_div: 1,
-                        },
-                        &mut root,
-                    )
-                    .await
-                    .expect("scan selected compacted object");
-                assert_eq!(
-                    attempts(global_metrics().report().await) - before,
-                    u64::from(select_prob != 0),
-                    "selected compacted object must reach {scan_mode:?} admission with divisor {select_prob}"
-                );
-                let total = scanner.new_cache.size_recursive("bucket").expect("usage root");
-                assert_eq!(total.objects, 1);
-                assert_eq!(total.versions, 1);
-            })
-            .await;
-        }
-    }
-}
-
-#[tokio::test(start_paused = true)]
-#[serial]
-async fn scanner_cancellation_interrupts_folder_throttle() {
-    use futures::{FutureExt, poll};
-    let (mut scanner, temp_dir) = build_test_scanner().await;
-    let _guard = TestGuard::new(0, 0, &mut scanner, temp_dir);
-    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Slowest);
-    let previous_idle = crate::sleeper::SCANNER_IDLE_MODE.swap(true, std::sync::atomic::Ordering::Relaxed);
-    let ctx = CancellationToken::new();
-    let mut root = DataUsageEntry::default();
-    let mut scan = scanner
-        .scan_folder(
-            ctx.clone(),
-            CachedFolder {
-                name: "bucket".to_string(),
-                parent: None,
-                object_heal_prob_div: 1,
-            },
-            &mut root,
-        )
-        .boxed();
-    assert!(poll!(scan.as_mut()).is_pending(), "scan should be waiting in its folder throttle");
-    ctx.cancel();
-    let outcome = scan.now_or_never();
-    crate::sleeper::SCANNER_IDLE_MODE.store(previous_idle, std::sync::atomic::Ordering::Relaxed);
-    assert!(
-        matches!(outcome, Some(Err(_))),
-        "cancellation must finish without advancing the sleep clock"
-    );
-}
-
-#[test]
-#[serial]
-fn scanner_zero_directory_cycle_keeps_rescanning_enabled() {
-    temp_env::with_var(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, Some("0"), || {
-        for cycle in 0..32 {
-            assert!(
-                hash_path("bucket/object").mod_(cycle, data_usage_update_dir_cycles()),
-                "zero must not leave compacted usage stale forever"
-            );
-        }
-    });
-}
-
 #[tokio::test]
 #[serial]
 async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
@@ -2470,7 +2153,7 @@ async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
         .await
         .expect("cached metadata failure must still stop erasure data directory descent");
 
-    assert_eq!(retry_into.failed_objects, 1, "cached failure must remain visible in each snapshot");
+    assert_eq!(retry_into.failed_objects, 0, "cached failure should not be counted twice");
     assert!(!retry_budget.budget_elapsed());
     assert_eq!(retry_budget.reason(), None);
 
@@ -2595,7 +2278,7 @@ async fn test_scan_folder_missing_xl_meta_stops_erasure_data_dir_descent() {
         .await
         .expect("cached missing metadata must still stop erasure data directory descent");
 
-    assert_eq!(retry_into.failed_objects, 1, "cached failure must remain visible in each snapshot");
+    assert_eq!(retry_into.failed_objects, 0, "cached failure should not be counted twice");
     assert!(!retry_budget.budget_elapsed());
     assert_eq!(retry_budget.reason(), None);
 }

docs/architecture/README.md

@@ -48,7 +48,7 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 | [config-model-boundary-adr.md](config-model-boundary-adr.md) | touching the server-config model (`Config`, `KV`, `KVS`) or its persistence, or asking which crate owns which part of server configuration |
 | [admin-route-action-snapshot.md](admin-route-action-snapshot.md) | adding, moving, or re-authorizing an admin route and needing to know where the route → handler → `AdminAction` contract is enforced |
 | [kms-bulk-rekey-contract.md](kms-bulk-rekey-contract.md) | changing the bulk envelope re-wrap sweep, its admin endpoints, the re-wrap primitive, or which objects a rekey may touch |
-| [remote-credential-sealing-adr.md](remote-credential-sealing-adr.md) | adding, reading, or persisting a stored remote credential (replication target, remote tier, on-demand migration source), or needing the sealed-envelope format and its mixed-version rules |
+| [remote-credential-sealing-adr.md](remote-credential-sealing-adr.md) | adding, reading, or persisting a stored remote credential (replication target, remote tier, on-demand migration source), or needing the sealed-envelope format, the mixed-version rules, or the reason this is worth doing in one deployment and not in another |
 | [tier-stats-contract.md](tier-stats-contract.md) | changing what `GET /rustfs/admin/v3/tier-stats` returns, adding a tier accounting source, or wiring a metric to a remote tier request |
 
 ## Support and compatibility matrices (release-facing, keep current)

docs/architecture/remote-credential-sealing-adr.md

@@ -1,52 +1,151 @@
 # Remote Credential Sealing ADR
 
-**Use this when:** you add, read, or persist a stored remote credential — a replication target, a remote tier, or an on-demand migration source — or you need the sealed-envelope format, its fail-closed rules, and the mixed-version compatibility matrix.
-**Source of truth:** the three stores that hold remote credentials today — `BUCKET_TARGETS_FILE` and `BUCKET_ON_DEMAND_MIGRATION_CONFIG` in `crates/ecstore/src/bucket/metadata.rs`, and `TIER_CONFIG_FILE` in `crates/ecstore/src/services/tier/tier.rs` — plus the consumers `crates/ecstore/src/bucket/bucket_target_sys.rs`, `crates/ecstore/src/services/tier/tier.rs`, and `crates/ecstore/src/bucket/on_demand_migration/config.rs`.
+**Use this when:** you add, read, or persist a stored remote credential — a replication target, a remote tier, or an on-demand migration source — or you need the sealed-envelope format, the mixed-version rules, or the reason this is worth doing in one deployment and not in another.
+**Source of truth:** the three stores that hold remote credentials — `BUCKET_TARGETS_FILE` and `BUCKET_ON_DEMAND_MIGRATION_CONFIG` in `crates/ecstore/src/bucket/metadata.rs`, `TIER_CONFIG_FILE` in `crates/ecstore/src/services/tier/tier.rs` — the shared envelope in `crates/ecstore/src/bucket/sealed_credentials.rs`, the consumers `crates/ecstore/src/bucket/bucket_target_sys.rs`, `crates/ecstore/src/services/tier/tier_config.rs` and `crates/ecstore/src/bucket/on_demand_migration/config.rs`, and the backend properties in [../operations/kms-backend-security.md](../operations/kms-backend-security.md).
+
+## Recommendation
+
+Seal the credentials, ship the write side off by default, and claim a security benefit only for deployments running the Vault Transit or AWS KMS backend — everywhere else recommend full-disk encryption and the fail-closed parse fix below, which cost no code and cover strictly more.
+
+## Whether encryption buys anything here
+
+This decides the whole question, so it comes before the design. Sealing converts "read the drives" into "read the drives **and** hold an authenticated path to the key". How much that is worth depends entirely on the KMS backend, and [../operations/kms-backend-security.md](../operations/kms-backend-security.md) is explicit about the difference.
+
+| Backend | Where the key that unwraps these credentials lives | What sealing is worth |
+|---|---|---|
+| Vault Transit, AWS KMS | Inside Vault or AWS; only ciphertext ever leaves | Real. An offline copy of `.rustfs.sys` is inert. Each unwrap is a live authenticated call that is logged, rate-limitable and revocable, and revoking the node's identity retroactively protects every copy already taken |
+| Vault KV2 | In Vault KV v2, Base64-encoded, not wrapped | Thin. The referenced document states that KV read access is equivalent to holding the master keys, so the boundary is the Vault ACL on the key prefix — worth something only when that ACL is genuinely narrower than access to the drives, and worth nothing against anyone holding both |
+| Local, Static | In `key_dir` on the node's own filesystem, or in the process environment | Close to nothing. Whoever reads the drives on a node usually reads the host too. The only gap it covers is media taken away from the host — the same gap full-disk encryption covers better |
+
+Two honest limits hold on every backend. Sealing is **not** a defense against code execution on a node: the sealer runs in-process on every node that has to build a remote client, so an attacker at that level asks it to unseal and gets the plaintext. And it is not a defense against an authorized admin, because an admin who can rewrite a target can point it at a remote they control instead of reading the old secret.
+
+What it does remove is the media-level read: a decommissioned or RMA'd drive, a drive-level backup or volume snapshot, a host path exposed by a bad mount, a copy of a drive taken for support. That threat is real, and it is the only one this design addresses.
 
 ## Decision
 
 Remote credentials are sealed **per field, into an added field, behind one shared seam**, and ECStore reaches KMS through an installed hook rather than a crate dependency.
 
-1. **One seam, three consumers.** `BucketTargetSys`, `TierConfigMgr`, and `OnDemandMigrationSys` seal and unseal through a single ECStore-owned envelope type. No consumer talks to KMS, and no consumer defines its own ciphertext layout.
-2. **Only secret material is sealed.** `secret_key` and `session_token` are sealed. Endpoint, region, ARN, bucket, prefixes, path style, TLS flags, and the custom CA bundle stay in clear text: they are needed for validation, listing, and support diagnosis, and none of them is a secret.
-3. **Sealed material lives in an added field, never in place of the plaintext field.** A record carries either the plaintext field or the sealed field. A reader that does not understand the sealed field therefore finds the credential *absent* rather than finding a ciphertext string it would sign requests with.
-4. **Unsealing happens at client construction, not at parse time.** `build_remote_s3_client` in `crates/ecstore/src/bucket/remote_s3_client.rs` is the single point that needs plaintext, so admin reads, listings, validation, and status paths never call KMS.
+1. **One seam, three consumers.** `BucketTargetSys`, `TierConfigMgr` and `OnDemandMigrationSys` seal and unseal through `crates/ecstore/src/bucket/sealed_credentials.rs`. No consumer talks to KMS, and no consumer defines its own ciphertext layout.
+2. **Only secret material is sealed,** and the enumeration comes from the redaction code, not from a pair of field names — see [Which fields are sealed](#which-fields-are-sealed). Endpoint, region, ARN, bucket, prefixes, path style, TLS flags and the custom CA bundle stay in clear text: they are needed for validation, listing and support diagnosis, and none of them is a secret.
+3. **Sealed material lives in an added field, and the plaintext field is emptied rather than removed.** A reader that does not understand the sealed field must see a credential that is *present and empty*, so it takes a missing-credential path rather than a parse failure. Removing the field instead is what turns this design into an outage; [Compatibility](#compatibility-per-store-because-the-three-differ) explains why.
+4. **Unsealing happens at client construction, not at parse time.** `build_remote_s3_client` in `crates/ecstore/src/bucket/remote_s3_client.rs` is the only place that needs plaintext, so admin reads, listings, validation and status paths never call KMS — and a KMS outage never changes which targets or tiers *exist*.
+
+## What is stored today, and where
+
+Two of the three are not files at all. `bucket-targets.json` and `on-demand-migration.json` are named sub-configurations inside one msgpack blob per bucket, and only the tier configuration is its own object.
+
+| Store | Reached as | Actually persisted at | Written by | Container |
+|---|---|---|---|---|
+| Replication and ILM targets | `BUCKET_TARGETS_FILE` | `BucketMetadata::bucket_targets_config_json`, msgpack field `BucketTargetsConfigJSON` | `BucketMetadata::update_config`, then `BucketMetadata::save_with_store`; `crates/ecstore/src/bucket/metadata_sys.rs` serializes the update under a transaction lock | `{BUCKET_META_PREFIX}/{bucket}/{BUCKET_METADATA_FILE}` in `RUSTFS_META_BUCKET` (`crates/ecstore/src/disk/mod.rs`) |
+| On-demand migration source | `BUCKET_ON_DEMAND_MIGRATION_CONFIG` | `BucketMetadata::on_demand_migration_config_json`, msgpack field `OnDemandMigrationConfigJSON` | same path; `update_config` additionally refuses a blob this build cannot parse | same blob as above |
+| Remote tiers | `TIER_CONFIG_FILE` | its own object, a four-byte `TIER_CONFIG_FORMAT` / `TIER_CONFIG_VERSION` header followed by an `rmp_serde` payload of `ExternalTierConfigMgr` | `TierConfigMgr` through `encode_external_tiering_config_blob`, under `tier_config_lock_path` | `tier_config_path` under `CONFIG_PREFIX` in `RUSTFS_META_BUCKET` |
+
+The consequence of the first two sharing a blob is that any change to how that blob parses has a blast radius covering policy, lifecycle, versioning, object lock and everything else in `BucketMetadata` — not just credentials.
+
+## The at-rest boundary as it stands
+
+Three things hold the line today, and all three keep working whether or not sealing ships.
+
+- **The reserved bucket.** `RUSTFS_META_BUCKET` is `.rustfs.sys`; `is_reserved_or_invalid_bucket` keeps it off the S3 surface, and the admin inspect archive in `rustfs/src/admin/handlers/inspect_archive.rs` runs its request through a strict bucket-name check that a dot-prefixed reserved name does not pass.
+- **Admin authorization** on every route that can read or write one of the three configurations.
+- **Redaction on every read path.** `BucketTarget::redacted_credentials` and the `Debug` for `Credentials` in `crates/ecstore/src/bucket/target/bucket_target.rs`, used by the remote-target listing in `rustfs/src/admin/handlers/replication.rs` and by the bucket-metadata export in `rustfs/src/admin/handlers/bucket_meta.rs`; `TierConfig::redacted` in `crates/ecstore/src/services/tier/tier_config.rs`, which is also what that type's `Clone` and `Debug` do; and `SourceCredentials::redacted` in `crates/ecstore/src/bucket/on_demand_migration/config.rs`, used by `rustfs/src/admin/handlers/on_demand_migration.rs`.
+
+So no API returns a stored secret. The bytes are reachable by reading the drives, and that is the boundary sealing is proposed to move.
+
+## Which fields are sealed
+
+The authoritative list of what this codebase treats as secret is the redaction functions above, and it is wider than `secret_key` plus `session_token`.
+
+| Store | Sealed | Left in clear text although redacted |
+|---|---|---|
+| Targets | `Credentials::secret_key`, `Credentials::session_token` | — |
+| On-demand migration | `SourceCredentials::secret_key`, `SourceCredentials::session_token` | — |
+| Tiers | `secret_key` on each of the nine S3-family backends in `crates/ecstore/src/services/tier/tier_config.rs`, `TierAzure::sp_auth.client_secret`, and `TierGCS::creds` | `TierS3::aws_role_web_identity_token_file`, which is a path rather than a secret |
+
+`TierGCS::creds` carries a whole service-account key and is the largest single secret of the three stores; a design that sealed only fields literally named `secret_key` would leave it in clear text. `aws_role_web_identity_token_file` points at a file outside `.rustfs.sys`, so sealing it would protect nothing — and a tier configured that way stores no long-lived secret at all, which is the cheapest mitigation available and should be preferred where the remote supports it.
 
 ## Envelope format
 
-A versioned, self-describing record: envelope version, KMS key id, KMS key version, algorithm, nonce, and ciphertext. It is stored base64 in the two JSON stores and as raw bytes inside the msgpack payload of the tier blob; the tier blob's own `TIER_CONFIG_FORMAT` / `TIER_CONFIG_VERSION` header constants are unchanged, because the envelope carries its own version.
+`SealedCredential` in `crates/ecstore/src/bucket/sealed_credentials.rs`: envelope version, KMS key id, optional KMS key version, algorithm label, and the ciphertext produced by the sealer. It is stored base64 in the two JSON stores and as bytes alongside the tier payload. `SEALED_CREDENTIAL_VERSION` is checked by `SealedCredential::check_version` *before* the sealer is consulted, so an envelope from a newer build is refused here rather than inside a backend.
 
-The KMS encryption context binds each ciphertext to the record that owns it — store kind, owning bucket or tier name, and field name — so a ciphertext copied into another bucket, another tier, or another field fails to decrypt instead of silently authorizing a different remote.
+The encryption context binds each ciphertext to the record that owns it. `SealScope` renders store kind, owner (bucket name, tier name or target ARN) and field name into the context, so a ciphertext copied into another bucket, another tier or another field fails to decrypt instead of silently authorizing a different remote. Those context keys are part of the on-disk contract: changing one makes every existing ciphertext undecryptable.
+
+The envelope deliberately does **not** carry its own scope. A scope read out of the stored bytes would be attacker-controlled, and checking a ciphertext against a context it supplied itself proves nothing. The scope is always re-derived from where the ciphertext was found, which is also a constraint on any rewrap job — see [Rotation](#rotation).
 
 ## Why a hook instead of a dependency
 
-`crates/ecstore/Cargo.toml` has no `rustfs-kms` dependency, and adding one would invert the crate layering. The established shape is an `OnceLock` hook that ECStore defines and the binary installs at startup: `EVENT_DISPATCH_HOOK` in `crates/ecstore/src/services/event_notification.rs`, installed by `install_ecstore_event_dispatch_hook` in `rustfs/src/server/event.rs`, and `ON_DEMAND_MIGRATION_CONFIG_HOOK` in `crates/ecstore/src/bucket/on_demand_migration/config.rs`. Sealing uses the same shape, with the binary supplying an implementation backed by `get_global_kms_service_manager` in `crates/kms/src/service_manager.rs`.
+`crates/ecstore/Cargo.toml` has no `rustfs-kms` dependency, and adding one would invert the crate layering described in [crate-boundaries.md](crate-boundaries.md). The established shape is an `OnceLock` hook that ECStore defines and the binary installs at startup, as `EVENT_DISPATCH_HOOK` in `crates/ecstore/src/services/event_notification.rs` and `ON_DEMAND_MIGRATION_CONFIG_HOOK` in `crates/ecstore/src/bucket/on_demand_migration/config.rs` already do. `install_credential_sealer` follows it, and the binary supplies an implementation backed by `crates/kms/src/service_manager.rs`.
 
-## Compatibility matrix
+## Compatibility, per store, because the three differ
 
-| Stored form | Reader | Behavior |
-|---|---|---|
-| Plaintext (today's format) | Old node | Unchanged. |
-| Plaintext | New node | Read as plaintext, no KMS call. Carries a `RUSTFS_COMPAT_TODO` marker per [compat-cleanup-register.md](compat-cleanup-register.md). |
-| Sealed | New node, hook installed | Unsealed at client construction. |
-| Sealed | New node, no hook or decrypt failure | Typed error; the target, tier, or source is unusable and reports why. Never a default, an empty credential, or the ciphertext bytes. |
-| Sealed | Old node | The credential field is absent, so the old node fails closed on its existing "missing credentials" path. This is the migration hazard the rollout gate exists for. |
+The generic matrix is short: a plaintext record reads unchanged on any node; a sealed record reads on a new node with a sealer installed; a sealed record on a new node without one is a typed error and never a default. Everything difficult is in what an **old** node does, and the three stores behave differently enough that a single answer would be wrong.
 
-## Rollout gate
+| Store | Old node meets an added sealed field | Old node meets an emptied plaintext field | Verdict |
+|---|---|---|---|
+| Targets | Ignored. `BucketTarget` and `Credentials` do not use `deny_unknown_fields` | `Credentials` has no struct-level `serde(default)`, so a **missing** `secretKey` is a hard parse error for the whole document — but an **empty** one parses | Safe only if the plaintext field is emptied rather than removed |
+| On-demand migration | **Rejected.** `OnDemandMigrationConfig`, `SourceConfig` and `SourceCredentials` all carry `deny_unknown_fields`, so the whole configuration becomes unreadable, and `BucketMetadata::update_config` also refuses to persist it | Parses | Needs a reader-first release before any node writes the field |
+| Tiers | The payload is compact `rmp_serde`, which encodes structs positionally; an added field is an arity change a reader built for the previous struct cannot skip. `decode_external_tiering_config_blob` also rejects any `TIER_CONFIG_VERSION` it does not know | Parses | The sealed value must not be added to any struct inside the existing payload |
 
-Sealing is written only when KMS is configured **and** a module switch in `rustfs/src/module_switches.rs` is on, defaulting off in the release that introduces it. Reading sealed records is always supported; writing them is what waits. Operators enable the switch after every node in the cluster can read the format, and existing plaintext records are sealed by re-submitting the configuration through its admin API — this task ships no in-place migration sweep.
+Two of those rows are load-bearing enough to spell out.
+
+**Targets.** `BucketMetadata::parse_all_configs` responds to an unparseable `bucket-targets.json` by logging `bucket_metadata_parse_failed` and setting `bucket_target_config` to `BucketTargets::default()` — an empty target list. So on an old node a record whose `secretKey` was removed does not fail per target: **every target in that bucket disappears, replication stops, and no caller sees an error.** The raw bytes survive in the blob, so it is recoverable, but the silence is the hazard. Emptying the field instead of removing it avoids triggering it, and the substitution itself should be replaced by a retained parse failure before any of this ships — see [Prerequisites](#prerequisites-in-this-order).
+
+An emptied `secretKey` is not yet a clean local failure either. `build_remote_s3_client` raises `RemoteS3ClientError::MissingCredentials` only when the whole credentials object is absent, and `remote_sdk_credentials` passes an empty secret to the SDK, so today an emptied field signs a request that the remote rejects. That is loud rather than silent, and therefore acceptable as a floor, but the reader-first release should turn an empty access key or secret key into the same typed local error so the failure is attributable to this node instead of to the remote.
+
+**Tiers.** A format change to `tier-config.bin` takes out every tier at once, and tiers are not only a write-path concern: an object already transitioned to a tier cannot be read without that tier's configuration, so the failure reaches GETs of data that has been there for months. The sealed values therefore belong in a companion object under the same prefix, covered by the same `tier_config_lock_path`, keyed by tier name and field name, leaving `tier-config.bin` byte-shaped exactly as it is with an empty `SecretKey`. Putting the envelope *into* `SecretKey` was considered and rejected: an old node would sign requests with the ciphertext, producing remote 403s and ciphertext in signature-related logs, instead of taking its missing-credential path. Confirm the exact decode behaviour against the encode/decode tests in `crates/ecstore/src/services/tier/tier.rs` before writing a byte of the new layout, and do not bump `TIER_CONFIG_VERSION` until every node in the supported upgrade range reads it.
+
+**Downgrade** is the same event as "old node reads new bytes", with one addition: a node that has been downgraded keeps writing the old shape, so a configuration re-submitted through it loses the sealed field and returns to plaintext. That is a security regression, not a correctness one, and it is silent — which is another reason the write side is gated rather than defaulted on.
+
+## KMS unavailable: read time versus write time
+
+These two are not symmetric, and conflating them is how this design would cause an outage.
+
+**At write time** the answer is easy: sealing fails, the admin write is refused with the typed error, and nothing is persisted. A configuration is never stored with the secret dropped, and never stored in clear text after the operator asked for sealing. The cost is that configuration cannot be changed while the KMS is down, which is acceptable and visible.
+
+**At read time** the rule is that a credential which cannot be unsealed makes a remote *unusable*, never *absent*.
+
+- Because unsealing happens at `build_remote_s3_client`, a KMS outage does not change which targets or tiers exist. Listings, status and admin reads keep returning them; each attempt to use one fails with a typed, retryable error that names the KMS as the cause.
+- Startup must not treat "cannot unseal" as "no such tier". A tier whose credential is unavailable stays present in `TierConfigMgr`, so a GET of an object transitioned to it fails with a retryable error rather than presenting as missing data, and nothing re-drives a transition elsewhere. The same holds for a replication target: it stays configured and reports why it is not working.
+- **A write must refuse to rewrite a configuration it could not fully read.** This is the sharpest edge in the whole design. If a partially-unreadable configuration can be re-serialized from a partially-populated in-memory view, then a KMS outage plus one unrelated admin edit persists the configuration with the unreadable records dropped — and that is the only mechanism by which a target or tier really would disappear for good. Today's code does not have this hazard, because both stores keep raw bytes or fail the whole decode; any per-record sealed handling that skips undecodable records would introduce it.
 
 ## Rotation
 
-The envelope records the key id and key version it was wrapped under. Re-wrapping is the KMS side's job and follows [kms-bulk-rekey-contract.md](kms-bulk-rekey-contract.md); nothing in this design rotates, re-wraps, or expires a key on its own.
+The envelope records the key id and, when the backend reports one, the key version. Re-wrapping is the KMS side's job, follows [kms-bulk-rekey-contract.md](kms-bulk-rekey-contract.md) and rustfs/backlog#1637 and #1642, and nothing here rotates, re-wraps or expires a key on its own. Two properties make that division workable, and both are constraints on the rewrap job rather than on this design.
+
+- A rewrap must reproduce the encryption context, and the envelope does not carry it. The job must therefore reach a ciphertext **through its store** — enumerate targets, tiers and migration sources and derive the `SealScope` from the record's own position — rather than by scanning for envelope-shaped bytes.
+- `key_version` is absent for backends that report none. An absent version means "not known to be current", never "current"; a rewrap sweep must be able to act on it, and a completed sweep is evidence about scanned sources only, exactly as the referenced contract already says about key deletion.
 
 ## Fail-closed rules
 
-- A missing hook, a malformed envelope, an unknown envelope version, a failed decrypt, or an encryption-context mismatch is a typed error, per the AGENTS.md rule that required values return a typed error when absent or corrupt.
-- A seal failure fails the admin write. A configuration is never persisted with the secret dropped or left in clear text after the operator asked for sealing.
-- Redaction is unchanged and independent: admin responses keep returning `REDACTED`, and `Debug` implementations keep hiding secret fields whether or not the stored form is sealed.
-- Logs may carry the key id and envelope version. They never carry ciphertext, plaintext, or the encryption context's secret-adjacent values.
+- A missing sealer, a malformed envelope, an unknown envelope version, a failed decrypt or an encryption-context mismatch is a typed error, per the root `AGENTS.md` rule that a required value returns a typed error when it is absent or corrupt. `SealedCredentialError` has no variant that degrades to a default, an empty credential, or the raw ciphertext.
+- A seal failure fails the admin write.
+- Redaction is unchanged and independent: admin responses keep returning `REDACTED`, and `Debug` implementations keep hiding secret fields whether or not the stored form is sealed. `SealedCredential`'s own `Debug` prints the key id and a byte count, not the ciphertext.
+- Logs may carry the key id and the envelope version. They never carry ciphertext, plaintext, or an encryption-context value.
+- A sealed value never enters an equality or fingerprint comparison. `tier_config_fingerprint` hashes a tier configuration to decide whether an edit is a no-op, and `OnDemandMigrationConfig` derives `PartialEq`; a fresh nonce per seal would make every write look like a change and churn the tier driver cache. Compare plaintext configurations, then seal.
+
+## Alternatives considered
+
+**Do not encrypt; harden the existing boundary instead.** This is the strongest alternative, not a foil. Its parts: keep `.rustfs.sys` off every request surface, which already holds; make an unparseable `bucket-targets.json` fail closed instead of becoming an empty list, which is a bug fix worth doing regardless; prefer keyless credentials where the remote supports them, as `TierS3::aws_role_web_identity_token_file` already allows; and encrypt the drives, which removes the media threat completely, covers all three stores plus every other secret in `.rustfs.sys`, and costs no code. Against the media threat, full-disk encryption strictly dominates application-level sealing. Sealing wins only where the KMS is Transit or AWS **and** the operator wants each unwrap to be individually authenticated, logged and revocable — which is exactly the scope this ADR claims and no more.
+
+**Encrypt the whole blob, as MinIO does for its tier configuration.** Rejected. `tier-config.bin`'s header is what tells a reader the format, and a whole-blob ciphertext makes every tier unreadable whenever the KMS is unreachable; for the bucket metadata blob it would take policy, lifecycle, versioning and object lock down with the credential. Per-field sealing keeps the blast radius at one credential.
+
+**Keep the credential in the KMS and store only a reference.** Rejected. It makes the KMS the durability authority for configuration, adds a second lifecycle with its own orphans when a bucket or tier is deleted, and none of the supported backends is a general secret store — the backends documented in [../operations/kms-backend-security.md](../operations/kms-backend-security.md) manage keys, not arbitrary secrets.
+
+**Deterministic encryption so ciphertext is stable across writes.** Rejected. It weakens the encryption to make ciphertext comparable, and the thing that wanted comparable bytes — configuration-change detection — is correctly solved by comparing plaintext configurations before sealing.
+
+**Seal inside `TierConfig` rather than at the persistence boundary.** Rejected. That type's `Clone` is `redacted()`, so cloning drops secrets, and `tier_config_fingerprint` hashes the type; a nondeterministic sealed field inside it would be both lossy and churn-inducing.
+
+**Encrypt with a node-local key instead of the KMS.** Rejected. The key would sit on the same host as the data, so it removes nothing the reserved path does not already remove, and it creates key material that nothing rotates.
+
+## Prerequisites, in this order
+
+1. Make an unparseable `bucket-targets.json` fail closed in `BucketMetadata::parse_all_configs` instead of substituting `BucketTargets::default()`. This is independently correct and it is what keeps a later mistake from being silent.
+2. Make an empty access key or secret key a typed `RemoteS3ClientError` in `remote_sdk_credentials`, so an emptied plaintext field fails on this node rather than as a signature rejection at the remote.
+3. Ship a reader-first release: every store tolerates the sealed field and the emptied plaintext field, and nothing writes either. For on-demand migration this means relaxing `deny_unknown_fields` for exactly that field name; for tiers it means reading the companion object when present.
+4. Only then enable writing, gated on KMS being configured and on a module switch in `rustfs/src/module_switches.rs` that defaults off in the release introducing it. Operators turn it on once every node reads the format. Existing plaintext records convert by re-submitting the configuration through its admin API; this work ships no in-place migration sweep.
+
+Steps 1 through 3 each introduce a compatibility path that needs a `RUSTFS_COMPAT_TODO` marker and a matching entry in [compat-cleanup-register.md](compat-cleanup-register.md) when the code lands. This document adds neither, because the guard matches markers and register entries in both directions and an entry without a marker fails it.
 
 ## Non-goals
 
-Sealing the server config, IAM credentials, or object data keys; changing which principals may read a configuration; key material migration between KMS backends; and any at-rest protection when KMS is not configured — without KMS the stored form stays plaintext and the existing trust boundary (reserved bucket paths plus admin authorization) is unchanged.
+Sealing the server configuration, IAM credentials or object data keys; changing which principals may read a configuration; migrating key material between KMS backends; and any at-rest protection when KMS is not configured — without KMS the stored form stays plaintext and the boundary described above is unchanged.

docs/operations/scanner-benchmark-runbook.md

@@ -34,22 +34,6 @@ The `scanner` and `heal` subsystems are served by `GetConfigKVHandler` (`rustfs/
 
 ## Test Matrix
 
-### Deterministic regression checks
-
-Run the scanner regressions before collecting host-pressure measurements:
-
-```bash
-cargo nextest run -p rustfs-scanner --lib
-```
-
-Most tests in `crates/scanner/tests/lifecycle_integration_test.rs` are ignored in the default lane because they require serial execution. Run the scanner portion of the `ILM Integration (serial)` selection in `.github/workflows/ci.yml` with `-j1 --run-ignored all` as well; preserve its documented exclusions for known noncurrent transition/expiry failures.
-
-The folder regressions exercise real directory enumeration and metadata decoding. `scanner_failed_child_retains_usage_and_scans_healthy_sibling` replaces an enumerated directory before descent, so its I/O failure is reproducible without depending on Unix permission enforcement. `scanner_nested_metadata_failure_without_retry_cache_is_partial_then_recovers` checks fresh and inherited failure state with retry caching disabled, reuse of a partial compacted subtree, and recovery on the next selected directory cycle. Neither a failed subtree nor an expired retry ledger proves zero usage.
-
-`scanner_compacted_directory_keeps_aggressive_heal_and_bitrot_sampling` covers disabled, sub-interval, and exact-interval heal divisors in normal and deep modes. `scanner_cancellation_interrupts_folder_throttle` uses a paused clock to require immediate cooperative cancellation. `scanner_blocked_expiry_preserves_usage_replication_and_integrity_work` covers lifecycle enabled/disabled with pending replication, failed replication, and Legal Hold; retained bytes and integrity/replication inspection must survive blocked expiry.
-
-These checks complement the sampling and cancellation design in [MinIO's scanner implementation](https://github.com/minio/minio/blob/master/cmd/data-scanner.go), especially `scanDataFolder`, `folderScanner.scanFolder`, and `dynamicSleeper.Sleep`. Scanner admission counters prove that work reaches the admission boundary; they do not prove remote replication delivery or a completed shard repair. The deployment matrix below remains necessary for those claims and for measured CPU, memory, IOPS, and foreground-latency comparisons.
-
 Collect at least two runs on the same RustFS commit and the same workload. Keep hardware, commit, object count, object size, bucket count, scanner-enabled state, and foreground workload constant between runs.
 
 | Run | Purpose | Example scanner settings |

docs/operations/scanner-runtime-controls.md

@@ -70,9 +70,9 @@ These have no persistent key and are read from the environment only.
 | `RUSTFS_SCANNER_ENABLED` (deprecated alias `RUSTFS_ENABLE_SCANNER`) | `true` (`scanner_enabled_from_env`, `rustfs/src/module_switches.rs`) | Starts the data scanner at all. The heal manager is initialized whenever heal or scanner is enabled, because scanner-produced heal candidates need a consumer. |
 | `RUSTFS_SCANNER_ALERT_COOLDOWN_SECS` | `86400` (`DEFAULT_SCANNER_ALERT_COOLDOWN_SECS`, `scanner_folder.rs`) | Per-(kind, bucket, object) cooldown between S3 excess-alert events; `0` emits every cycle. See [Scanner Excess Alerts](scanner-excess-alerts.md). |
 | `RUSTFS_SCANNER_DEEP_VERIFY_COOLDOWN_SECS` | `60` (`DEFAULT_SCANNER_DEEP_VERIFY_COOLDOWN_SECS`, `scanner_folder.rs`) | Objects modified within this window are skipped by deep (bitrot) verification in the current cycle. |
-| `RUSTFS_HEAL_OBJECT_SELECT_PROB` | `1024` (`DEFAULT_HEAL_OBJECT_SELECT_PROB`, `scanner_folder.rs`) | Sampling divisor for scanner-originated heal checks: roughly one object in N per cycle is selected for a low-priority heal check. `0` disables sampled checks. When N is smaller than the compacted-directory interval, every object in a selected directory is eligible; compaction must not round the sampling probability to zero. |
-| `RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES` | `16` (`DATA_USAGE_UPDATE_DIR_CYCLES`, `scanner_folder.rs`) | Every N cycles a compacted directory is re-descended instead of reusing its cached usage. `1` forces re-descent every cycle (used by lifecycle e2e lanes); `0` is normalized to `1`. |
-| `RUSTFS_DATA_USAGE_FAILED_OBJECT_TTL_SECS` | `86400` (`DEFAULT_FAILED_OBJECT_TTL_SECS`, `scanner_folder.rs`) | Retention of per-bucket failed-object retry entries in the usage cache. `0` disables and clears the retry cache; it does not allow failed scans to publish complete usage. Cached failures remain visible in each partial snapshot without extending their retry deadline. |
+| `RUSTFS_HEAL_OBJECT_SELECT_PROB` | `1024` (`DEFAULT_HEAL_OBJECT_SELECT_PROB`, `scanner_folder.rs`) | Sampling divisor for scanner-originated heal checks: roughly one object in N per cycle is selected for a low-priority heal check. |
+| `RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES` | `16` (`DATA_USAGE_UPDATE_DIR_CYCLES`, `scanner_folder.rs`) | Every N cycles a compacted directory is re-descended instead of reusing its cached usage. `1` forces re-descent every cycle (used by lifecycle e2e lanes). |
+| `RUSTFS_DATA_USAGE_FAILED_OBJECT_TTL_SECS` | `86400` (`DEFAULT_FAILED_OBJECT_TTL_SECS`, `scanner_folder.rs`) | Retention of per-bucket failed-object entries in the usage cache. |
 | `RUSTFS_DATA_USAGE_FAILED_OBJECTS_MAX` | `10000` (`DEFAULT_FAILED_OBJECTS_MAX`, `scanner_folder.rs`) | Cap on retained failed-object entries per bucket. |
 
 ### Cycle budgets and cadence

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