docs(architecture): settle remote credential sealing threat model (#7168)

* docs(architecture): settle remote credential sealing threat model

* docs(architecture): index sealing ADR threat-model scope

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/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/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