test(tier): remove redundant intent clones

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

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

@@ -25,6 +25,7 @@ use super::{
     manual_transition_job, tier_delete_journal, transition_transaction,
 };
 use crate::error::{Error, Result};
+use crate::services::tier::tier_probe_intent;
 
 pub(crate) const ILM_META_PREFIX: &str = "ilm";
 const ILM_META_OBJECT_PREFIX: &str = "ilm/";
@@ -35,6 +36,7 @@ pub(crate) enum DurableIlmRecordKind {
     TierDeleteJournal,
     TierDeleteDispatchManifest,
     TransitionTransaction,
+    TierProbeIntent,
     ManualTransitionJob,
     ManualTransitionScope,
     ManualTransitionTask,
@@ -73,6 +75,12 @@ pub(crate) const TRANSITION_TRANSACTION_NAMESPACE: DurableIlmNamespace = Durable
     max_record_size: transition_transaction::MAX_TRANSITION_TRANSACTION_SIZE,
     kind: DurableIlmRecordKind::TransitionTransaction,
 };
+pub(crate) const TIER_PROBE_INTENT_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace {
+    name: "tier-probe-intent",
+    prefix: tier_probe_intent::TIER_PROBE_INTENT_RECORD_PREFIX,
+    max_record_size: tier_probe_intent::MAX_TIER_PROBE_INTENT_SIZE,
+    kind: DurableIlmRecordKind::TierProbeIntent,
+};
 pub(crate) const MANUAL_TRANSITION_JOB_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace {
     name: "manual-transition-job",
     prefix: "ilm/manual-transition/jobs",
@@ -98,11 +106,12 @@ pub(crate) const MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE: DurableIlmNamespace
     kind: DurableIlmRecordKind::ManualTransitionWorkerResult,
 };
 
-pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 8] = [
+pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 9] = [
     TIER_DELETE_JOURNAL_NAMESPACE,
     TIER_DELETE_JOURNAL_V6_NAMESPACE,
     TIER_DELETE_DISPATCH_MANIFEST_NAMESPACE,
     TRANSITION_TRANSACTION_NAMESPACE,
+    TIER_PROBE_INTENT_NAMESPACE,
     MANUAL_TRANSITION_JOB_NAMESPACE,
     MANUAL_TRANSITION_SCOPE_NAMESPACE,
     MANUAL_TRANSITION_TASK_NAMESPACE,
@@ -200,6 +209,15 @@ pub(crate) enum DurableIlmRecordCheckpoint {
         revision: u64,
         state: transition_transaction::TransitionTransactionState,
     },
+    TierProbeIntent {
+        content_sha256: String,
+        identity_sha256: String,
+        remote_version_sha256: String,
+        remote_version_known: bool,
+        owner_fence_sha256: String,
+        revision: u64,
+        state: tier_probe_intent::TierProbeIntentState,
+    },
     ManualTransitionJob {
         content_sha256: String,
         identity_sha256: String,
@@ -232,6 +250,7 @@ impl DurableIlmRecordCheckpoint {
             | Self::TierDeleteDispatchManifest { content_sha256, .. }
             | Self::TierDeleteDispatchParent { content_sha256, .. }
             | Self::TransitionTransaction { content_sha256, .. }
+            | Self::TierProbeIntent { content_sha256, .. }
             | Self::ManualTransitionJob { content_sha256, .. }
             | Self::ManualTransitionScope { content_sha256, .. }
             | Self::ManualTransitionTask { content_sha256 }
@@ -421,6 +440,32 @@ impl DurableIlmRecordCheckpoint {
                         .is_some_and(|expected_revision| *next_revision == expected_revision)
                     && (!previous_remote_version_known || previous_remote_version == next_remote_version)
             }
+            (
+                Self::TierProbeIntent {
+                    identity_sha256: previous_identity,
+                    remote_version_sha256: previous_remote_version,
+                    remote_version_known: previous_remote_version_known,
+                    owner_fence_sha256: previous_owner_fence,
+                    revision: previous_revision,
+                    state: previous_state,
+                    ..
+                },
+                Self::TierProbeIntent {
+                    identity_sha256: next_identity,
+                    remote_version_sha256: next_remote_version,
+                    owner_fence_sha256: next_owner_fence,
+                    revision: next_revision,
+                    state: next_state,
+                    ..
+                },
+            ) => {
+                previous_identity == next_identity
+                    && previous_owner_fence == next_owner_fence
+                    && next_revision
+                        .checked_sub(*previous_revision)
+                        .is_some_and(|distance| distance == 1 && tier_probe_state_reaches(*previous_state, *next_state, distance))
+                    && (!previous_remote_version_known || previous_remote_version == next_remote_version)
+            }
             (
                 Self::ManualTransitionJob {
                     content_sha256: previous_content,
@@ -500,6 +545,14 @@ impl DurableIlmRecordCheckpoint {
     /// after the exact terminal ETag and terminal receipt were committed, to
     /// purge older object versions exposed by that deletion.
     pub(crate) fn is_predecessor_of_terminal(&self, terminal: &Self) -> bool {
+        if let Self::TierProbeIntent { state, .. } = terminal
+            && !matches!(
+                state,
+                tier_probe_intent::TierProbeIntentState::AbortedNoRemote | tier_probe_intent::TierProbeIntentState::Completed
+            )
+        {
+            return false;
+        }
         if self == terminal || self.validate_successor(terminal).is_ok() {
             return true;
         }
@@ -568,6 +621,37 @@ impl DurableIlmRecordCheckpoint {
                         }
                     })
             }
+            (
+                Self::TierProbeIntent {
+                    identity_sha256: previous_identity,
+                    remote_version_sha256: previous_remote_version,
+                    remote_version_known: previous_remote_version_known,
+                    owner_fence_sha256: previous_owner_fence,
+                    revision: previous_revision,
+                    state: previous_state,
+                    ..
+                },
+                Self::TierProbeIntent {
+                    identity_sha256: terminal_identity,
+                    remote_version_sha256: terminal_remote_version,
+                    owner_fence_sha256: terminal_owner_fence,
+                    revision: terminal_revision,
+                    state: terminal_state,
+                    ..
+                },
+            ) => {
+                previous_identity == terminal_identity
+                    && previous_owner_fence == terminal_owner_fence
+                    && matches!(
+                        terminal_state,
+                        tier_probe_intent::TierProbeIntentState::AbortedNoRemote
+                            | tier_probe_intent::TierProbeIntentState::Completed
+                    )
+                    && terminal_revision
+                        .checked_sub(*previous_revision)
+                        .is_some_and(|distance| tier_probe_state_reaches(*previous_state, *terminal_state, distance))
+                    && (!previous_remote_version_known || previous_remote_version == terminal_remote_version)
+            }
             _ => false,
         }
     }
@@ -606,6 +690,23 @@ fn transition_state_distance(
     }
 }
 
+fn tier_probe_state_reaches(
+    from: tier_probe_intent::TierProbeIntentState,
+    to: tier_probe_intent::TierProbeIntentState,
+    revision_distance: u64,
+) -> bool {
+    use tier_probe_intent::TierProbeIntentState::{AbortedNoRemote, CleanupPending, Completed, UploadOutcomeUnknown, Uploaded};
+
+    match (from, to) {
+        (UploadOutcomeUnknown, Uploaded | CleanupPending | AbortedNoRemote) => revision_distance == 1,
+        (UploadOutcomeUnknown, Completed) => matches!(revision_distance, 2 | 3),
+        (Uploaded, CleanupPending) => revision_distance == 1,
+        (Uploaded, Completed) => revision_distance == 2,
+        (CleanupPending, Completed) => revision_distance == 1,
+        _ => false,
+    }
+}
+
 fn manual_job_state_reaches(
     from: manual_transition_job::ManualTransitionJobState,
     to: manual_transition_job::ManualTransitionJobState,
@@ -1082,6 +1183,42 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
                 },
             )
         }
+        DurableIlmRecordKind::TierProbeIntent => {
+            let probe_id = tier_probe_intent::tier_probe_intent_id_from_record_object_name(path)
+                .map_err(|err| Error::other(err.to_string()))?;
+            let intent =
+                tier_probe_intent::TierProbeIntent::decode(probe_id, data).map_err(|err| Error::other(err.to_string()))?;
+            let canonical =
+                tier_probe_intent::tier_probe_intent_record_object_name(probe_id).map_err(|err| Error::other(err.to_string()))?;
+            if canonical != path {
+                return Err(Error::other("tier probe intent path is not canonical"));
+            }
+            let identity_sha256 = checkpoint_hash(&(
+                intent.probe_id,
+                &intent.operation,
+                &intent.tier_name,
+                intent.destination_id,
+                &intent.probe_object,
+                &intent.creator_id,
+                intent.creator_epoch,
+                intent.created_at_unix_nanos,
+            ))?;
+            let remote_version_sha256 = checkpoint_hash(&intent.remote_version)?;
+            let owner_fence_sha256 = checkpoint_hash(&intent.owner)?;
+            (
+                "probe_id",
+                probe_id.to_string(),
+                DurableIlmRecordCheckpoint::TierProbeIntent {
+                    content_sha256,
+                    identity_sha256,
+                    remote_version_sha256,
+                    remote_version_known: !intent.remote_version.is_unknown(),
+                    owner_fence_sha256,
+                    revision: intent.revision,
+                    state: intent.state,
+                },
+            )
+        }
         DurableIlmRecordKind::ManualTransitionJob => {
             let job_id = manual_transition_job::manual_transition_job_id_from_record_object_name(path)
                 .map_err(|err| Error::other(err.to_string()))?;
@@ -1237,6 +1374,102 @@ mod tests {
         }
     }
 
+    fn tier_probe_intent_fixture() -> tier_probe_intent::TierProbeIntent {
+        let probe_id = Uuid::parse_str("36e2220e-9ad2-495b-b3bc-c4d2caf70a31").expect("fixture uuid should parse");
+        tier_probe_intent::TierProbeIntent {
+            probe_id,
+            revision: 1,
+            state: tier_probe_intent::TierProbeIntentState::UploadOutcomeUnknown,
+            operation: tier_probe_intent::TierProbeOperationIdentity::Verify {
+                config_etag: "config-etag".to_string(),
+                backend_identity: [1; 32],
+            },
+            tier_name: "COLD-A".to_string(),
+            destination_id: [1; 32],
+            probe_object: tier_probe_intent::tier_probe_object_name(probe_id),
+            creator_id: "node-a".to_string(),
+            creator_epoch: Uuid::parse_str("76746062-c05a-40b7-9e38-d2722d7e0332").expect("fixture creator epoch should parse"),
+            created_at_unix_nanos: 1_780_000_000_000_000_000,
+            owner: tier_probe_intent::TierProbeOwnerFence {
+                owner_id: "node-a".to_string(),
+                owner_epoch: Uuid::parse_str("76746062-c05a-40b7-9e38-d2722d7e0332").expect("fixture owner epoch should parse"),
+                not_after_unix_nanos: 1_780_000_900_000_000_000,
+            },
+            remote_version: tier_probe_intent::TierProbeRemoteVersion::default(),
+        }
+    }
+
+    fn tier_probe_checkpoint(intent: &tier_probe_intent::TierProbeIntent) -> DurableIlmRecordCheckpoint {
+        let path =
+            tier_probe_intent::tier_probe_intent_record_object_name(intent.probe_id).expect("tier probe path should build");
+        let encoded = intent.encode().expect("tier probe intent should encode");
+        let namespace = classify_durable_ilm_record(&path)
+            .expect("tier probe namespace should classify")
+            .expect("tier probe intent should be durable");
+        assert_eq!(namespace, &TIER_PROBE_INTENT_NAMESPACE);
+        validate_durable_ilm_record(&path, &encoded)
+            .expect("tier probe intent should validate")
+            .checkpoint
+    }
+
+    #[test]
+    fn tier_probe_intent_checkpoint_tracks_exact_monotonic_generations() {
+        let initial_intent = tier_probe_intent_fixture();
+        let initial = tier_probe_checkpoint(&initial_intent);
+
+        let mut uploaded_intent = initial_intent;
+        uploaded_intent
+            .advance(
+                tier_probe_intent::TierProbeIntentState::Uploaded,
+                tier_probe_intent::TierProbeRemoteVersion::versioned("opaque-v1"),
+            )
+            .expect("uploaded state should advance");
+        let uploaded = tier_probe_checkpoint(&uploaded_intent);
+        initial
+            .validate_successor(&uploaded)
+            .expect("durable receipt may adopt the exact uploaded generation");
+
+        let mut cleanup_intent = uploaded_intent.clone();
+        cleanup_intent
+            .advance(
+                tier_probe_intent::TierProbeIntentState::CleanupPending,
+                uploaded_intent.remote_version.clone(),
+            )
+            .expect("cleanup state should advance");
+        let cleanup = tier_probe_checkpoint(&cleanup_intent);
+        uploaded
+            .validate_successor(&cleanup)
+            .expect("durable receipt may adopt the exact cleanup generation");
+
+        let mut completed_intent = cleanup_intent.clone();
+        completed_intent
+            .advance(tier_probe_intent::TierProbeIntentState::Completed, cleanup_intent.remote_version.clone())
+            .expect("completed state should advance");
+        let completed = tier_probe_checkpoint(&completed_intent);
+        cleanup
+            .validate_successor(&completed)
+            .expect("durable receipt may adopt the exact terminal generation");
+        assert!(
+            initial.is_predecessor_of_terminal(&completed),
+            "terminal cleanup must recognize the full acknowledged-PUT path"
+        );
+        assert!(
+            initial.validate_successor(&completed).is_err(),
+            "ordinary receipt advancement must not skip intermediate generations"
+        );
+        assert!(
+            !initial.is_predecessor_of_terminal(&uploaded),
+            "a nonterminal generation must not be accepted as terminal proof"
+        );
+
+        let mut rebound = uploaded_intent;
+        rebound.owner.owner_epoch = Uuid::new_v4();
+        assert!(
+            rebound.encode().is_err(),
+            "dormant v1 must reject owner takeover before producing a checkpoint"
+        );
+    }
+
     #[test]
     fn tier_delete_dispatch_manifest_namespace_validates_monotonic_branches() {
         use tier_delete_journal::TierDeleteDispatchManifestState::{Aborted, Aborting, Completed, DispatchAuthorized, Preparing};

crates/ecstore/src/bucket/sealed_credentials.rs

@@ -201,12 +201,6 @@ pub async fn unseal_secret(sealed: &SealedCredential, scope: &SealScope) -> Resu
 mod tests {
     use super::*;
     use parking_lot::Mutex;
-    use std::collections::BTreeMap;
-
-    fn encode_context(context: &HashMap<String, String>) -> String {
-        let ordered = context.iter().collect::<BTreeMap<_, _>>();
-        serde_json::to_string(&ordered).expect("context serializes")
-    }
 
     /// Stands in for the KMS-backed sealer: records the context it was called
     /// with, and refuses a ciphertext presented under a different one.
@@ -220,7 +214,7 @@ mod tests {
         async fn seal(&self, plaintext: &str, scope: &SealScope) -> Result<SealedCredential, SealedCredentialError> {
             let context = scope.encryption_context();
             self.sealed_contexts.lock().push(context.clone());
-            let mut bound = encode_context(&context);
+            let mut bound = serde_json::to_string(&context).expect("context serializes");
             bound.push('|');
             bound.push_str(plaintext);
             Ok(SealedCredential {
@@ -237,7 +231,7 @@ mod tests {
                 .decode_to_vec(sealed.ct.as_bytes())
                 .map_err(|err| SealedCredentialError::Malformed(err.to_string()))?;
             let bound = String::from_utf8(raw).map_err(|err| SealedCredentialError::Malformed(err.to_string()))?;
-            let expected = encode_context(&scope.encryption_context());
+            let expected = serde_json::to_string(&scope.encryption_context()).expect("context serializes");
             bound
                 .strip_prefix(&expected)
                 .and_then(|rest| rest.strip_prefix('|'))

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

@@ -21,6 +21,7 @@ pub mod tier_gen;
 pub mod tier_handlers;
 pub(crate) mod tier_mutation_intent;
 pub mod tier_mutation_peer;
+pub(crate) mod tier_probe_intent;
 pub mod warm_backend;
 pub mod warm_backend_aliyun;
 pub mod warm_backend_azure;

crates/ecstore/src/store/init.rs

@@ -864,6 +864,11 @@ mod tests {
                 save_tier_mutation_intent_record, save_tier_mutation_intent_record_if_current,
             },
             tier_mutation_peer::{TierMutationPeerError, TierMutationPeerState, handle_tier_mutation_peer_request},
+            tier_probe_intent::{
+                TierProbeIntent, TierProbeIntentState, TierProbeOperationIdentity, TierProbeOwnerFence, TierProbeRemoteVersion,
+                delete_tier_probe_intent_record_if_current, load_tier_probe_intent_record,
+                save_tier_probe_intent_record_if_absent, save_tier_probe_intent_record_if_current,
+            },
             warm_backend::{TransitionCandidateProbe, WarmBackend},
         },
         set_disk::SetDiskTransitionUploadedCommitBarrier as TransitionUploadedCommitBarrier,
@@ -17196,6 +17201,147 @@ mod tests {
         assert!(matches!(err, Error::ConfigNotFound));
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn tier_probe_intent_store_enforces_create_cas_and_terminal_delete_preconditions() {
+        let temp_dir = tempfile::tempdir().expect("create temp store dir");
+        let (_ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-probe-intent-cas", &[4])).await;
+        let probe_id = uuid::Uuid::new_v4();
+        let creator_epoch = uuid::Uuid::new_v4();
+        let initial = TierProbeIntent {
+            probe_id,
+            revision: 1,
+            state: TierProbeIntentState::UploadOutcomeUnknown,
+            operation: TierProbeOperationIdentity::Verify {
+                config_etag: "config-etag".to_string(),
+                backend_identity: [1; 32],
+            },
+            tier_name: "COLD-A".to_string(),
+            destination_id: [1; 32],
+            probe_object: format!("rustfs-tier-probe-{probe_id}"),
+            creator_id: "node-a".to_string(),
+            creator_epoch,
+            created_at_unix_nanos: 1_780_000_000_000_000_000,
+            owner: TierProbeOwnerFence {
+                owner_id: "node-a".to_string(),
+                owner_epoch: creator_epoch,
+                not_after_unix_nanos: 1_780_000_900_000_000_000,
+            },
+            remote_version: TierProbeRemoteVersion::default(),
+        };
+
+        save_tier_probe_intent_record_if_absent(store.clone(), &initial)
+            .await
+            .expect("initial probe intent should persist with create-only semantics");
+        let duplicate = save_tier_probe_intent_record_if_absent(store.clone(), &initial)
+            .await
+            .expect_err("duplicate create must fail closed");
+        assert!(matches!(duplicate, Error::PreconditionFailed));
+
+        let observed_initial = load_tier_probe_intent_record(store.clone(), probe_id)
+            .await
+            .expect("initial probe intent should load with an ETag");
+        assert_eq!(observed_initial.intent(), &initial);
+
+        let nonterminal_delete = delete_tier_probe_intent_record_if_current(store.clone(), &observed_initial)
+            .await
+            .expect_err("nonterminal evidence must not be deleted");
+        assert!(nonterminal_delete.to_string().contains("must be terminal"));
+
+        let mut fabricated_current_intent = initial.clone();
+        fabricated_current_intent.tier_name = "COLD-B".to_string();
+        let mut fabricated_successor = fabricated_current_intent.clone();
+        fabricated_successor
+            .advance(
+                TierProbeIntentState::Uploaded,
+                TierProbeRemoteVersion::versioned(uuid::Uuid::new_v4().to_string()),
+            )
+            .expect("fabricated successor should be internally valid");
+        let fabricated_current = observed_initial.with_intent_for_test(fabricated_current_intent.clone());
+        let crossed_cas = save_tier_probe_intent_record_if_current(store.clone(), &fabricated_current, &fabricated_successor)
+            .await
+            .expect_err("a live ETag must not authorize a different caller record");
+        assert!(matches!(crossed_cas, Error::PreconditionFailed));
+        assert_eq!(
+            load_tier_probe_intent_record(store.clone(), probe_id)
+                .await
+                .expect("crossed CAS must retain the authoritative record")
+                .intent(),
+            &initial
+        );
+
+        let mut fabricated_terminal_intent = fabricated_current_intent;
+        fabricated_terminal_intent
+            .advance(TierProbeIntentState::AbortedNoRemote, TierProbeRemoteVersion::default())
+            .expect("fabricated terminal should be internally valid");
+        let fabricated_terminal = observed_initial.with_intent_for_test(fabricated_terminal_intent);
+        let crossed_delete = delete_tier_probe_intent_record_if_current(store.clone(), &fabricated_terminal)
+            .await
+            .expect_err("a live ETag must not delete for a different caller record");
+        assert!(matches!(crossed_delete, Error::PreconditionFailed));
+        assert_eq!(
+            load_tier_probe_intent_record(store.clone(), probe_id)
+                .await
+                .expect("crossed delete must retain the authoritative record")
+                .intent(),
+            &initial
+        );
+
+        let remote_version = TierProbeRemoteVersion::versioned(uuid::Uuid::new_v4().to_string());
+        let mut uploaded = observed_initial.intent().clone();
+        uploaded
+            .advance(TierProbeIntentState::Uploaded, remote_version.clone())
+            .expect("known PUT result should advance");
+        save_tier_probe_intent_record_if_current(store.clone(), &observed_initial, &uploaded)
+            .await
+            .expect("the matching initial ETag should admit one successor");
+
+        let stale_cas = save_tier_probe_intent_record_if_current(store.clone(), &observed_initial, &uploaded)
+            .await
+            .expect_err("a consumed ETag must not overwrite the current generation");
+        assert!(matches!(stale_cas, Error::PreconditionFailed));
+
+        let observed_uploaded = load_tier_probe_intent_record(store.clone(), probe_id)
+            .await
+            .expect("uploaded generation should load");
+        assert_eq!(observed_uploaded.intent(), &uploaded);
+        let mut cleanup = observed_uploaded.intent().clone();
+        cleanup
+            .advance(TierProbeIntentState::CleanupPending, remote_version.clone())
+            .expect("known candidate should become cleanup-pending");
+        save_tier_probe_intent_record_if_current(store.clone(), &observed_uploaded, &cleanup)
+            .await
+            .expect("cleanup generation should persist by exact ETag");
+
+        let observed_cleanup = load_tier_probe_intent_record(store.clone(), probe_id)
+            .await
+            .expect("cleanup generation should load");
+        let mut completed = observed_cleanup.intent().clone();
+        completed
+            .advance(TierProbeIntentState::Completed, remote_version)
+            .expect("exact cleanup should become terminal");
+        save_tier_probe_intent_record_if_current(store.clone(), &observed_cleanup, &completed)
+            .await
+            .expect("terminal generation should persist by exact ETag");
+
+        let stale_terminal = observed_cleanup.with_intent_for_test(completed.clone());
+        let stale_delete = delete_tier_probe_intent_record_if_current(store.clone(), &stale_terminal)
+            .await
+            .expect_err("a stale ETag must not delete terminal evidence");
+        assert!(matches!(stale_delete, Error::PreconditionFailed));
+
+        let observed_completed = load_tier_probe_intent_record(store.clone(), probe_id)
+            .await
+            .expect("terminal generation should remain after stale delete");
+        assert_eq!(observed_completed.intent(), &completed);
+        delete_tier_probe_intent_record_if_current(store.clone(), &observed_completed)
+            .await
+            .expect("the exact terminal ETag should delete the record");
+        assert!(matches!(load_tier_probe_intent_record(store, probe_id).await, Err(Error::ConfigNotFound)));
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]

crates/scanner/src/scanner_io.rs

@@ -105,6 +105,12 @@ struct DirtyUsageSnapshot {
     covers_all_pending: bool,
 }
 
+#[derive(Clone, Debug, Default)]
+pub(crate) struct ScannerBucketScanScope {
+    selected_buckets: Option<Arc<HashSet<String>>>,
+    baseline_scan_plan_digest: Option<DataUsageScanPlanDigest>,
+}
+
 pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool {
     matches!(err, StorageError::Io(io) if io.to_string().starts_with(SCANNER_METADATA_CORRUPT_ERROR))
 }
@@ -146,6 +152,7 @@ fn object_lock_config_enabled(config: &ObjectLockConfiguration) -> bool {
 pub struct ScannerBucketScanPlan {
     buckets: Vec<BucketInfo>,
     all_buckets: Arc<Vec<BucketInfo>>,
+    scope: ScannerBucketScanScope,
     digest: DataUsageScanPlanDigest,
     leader_epoch: u64,
     tier_registry_generation: u64,
@@ -732,6 +739,8 @@ mod dirty_usage;
 mod guards;
 mod io_cache;
 mod io_cycle;
+#[cfg(test)]
+use io_cache::{ScannerSetCacheGeneration, prepare_scoped_set_scan};
 pub(crate) use io_cycle::nsscanner_with_storage_status;
 mod io_disk;
 #[cfg(test)]

crates/scanner/src/scanner_io/io_cache.rs

@@ -14,6 +14,93 @@
 /// ScannerIOCache implementation for SetDisks: bucket ordering, worker fan-out, merge, and publish.
 use super::*;
 
+#[derive(Clone, Copy)]
+pub(super) struct ScannerSetCacheGeneration {
+    pub(super) want_cycle: u64,
+    pub(super) leader_epoch: u64,
+    pub(super) tier_registry_generation: u64,
+    pub(super) source: DataUsageCacheSource,
+    pub(super) scan_plan_digest: DataUsageScanPlanDigest,
+}
+
+pub(super) struct PreparedScopedSetScan {
+    pub(super) buckets: Vec<BucketInfo>,
+    pub(super) cache: DataUsageCache,
+}
+
+pub(super) fn prepare_scoped_set_scan(
+    old_cache: &DataUsageCache,
+    set_buckets: &[BucketInfo],
+    all_buckets: &[BucketInfo],
+    scope: &ScannerBucketScanScope,
+    generation: ScannerSetCacheGeneration,
+) -> Option<PreparedScopedSetScan> {
+    let (Some(selected_buckets), Some(baseline_scan_plan_digest)) = (&scope.selected_buckets, scope.baseline_scan_plan_digest)
+    else {
+        return None;
+    };
+    if selected_buckets.is_empty()
+        || !old_cache.info.snapshot_complete
+        || old_cache.info.last_update.is_none()
+        || old_cache.info.name != DATA_USAGE_ROOT
+        || old_cache.info.next_cycle > generation.want_cycle
+        || old_cache.info.leader_epoch != generation.leader_epoch
+        || old_cache.info.tier_registry_generation != Some(generation.tier_registry_generation)
+        || old_cache.info.source != Some(generation.source)
+        || old_cache.info.scan_plan_digest != Some(baseline_scan_plan_digest)
+        || old_cache.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
+        || old_cache.checked_flatten_complete_scope(DATA_USAGE_ROOT).is_none()
+    {
+        return None;
+    }
+
+    let mut cache = DataUsageCache {
+        info: DataUsageCacheInfo {
+            name: DATA_USAGE_ROOT.to_string(),
+            next_cycle: generation.want_cycle,
+            leader_epoch: generation.leader_epoch,
+            tier_registry_generation: Some(generation.tier_registry_generation),
+            source: Some(generation.source),
+            snapshot_complete: false,
+            scan_plan_digest: Some(generation.scan_plan_digest),
+            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+            lkg_snapshot_complete: true,
+            lkg_next_cycle: Some(old_cache.info.next_cycle),
+            lkg_last_update: old_cache.info.last_update,
+            lkg_leader_epoch: Some(old_cache.info.leader_epoch),
+            lkg_scan_plan_digest: old_cache.info.scan_plan_digest,
+            ..Default::default()
+        },
+        cache: HashMap::new(),
+    };
+    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+    let root_hash = crate::hash_path(DATA_USAGE_ROOT);
+    let mut current_bucket_names = HashSet::with_capacity(all_buckets.len());
+    for bucket in all_buckets {
+        if !current_bucket_names.insert(bucket.name.as_str()) {
+            return None;
+        }
+        if selected_buckets.contains(&bucket.name) {
+            cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+            continue;
+        }
+
+        let bucket_hash = crate::hash_path(&bucket.name);
+        old_cache.find(&bucket.name)?;
+        cache.copy_with_children(old_cache, &bucket_hash, &Some(root_hash.clone()));
+        cache.find(&bucket.name)?;
+    }
+
+    Some(PreparedScopedSetScan {
+        buckets: set_buckets
+            .iter()
+            .filter(|bucket| selected_buckets.contains(&bucket.name))
+            .cloned()
+            .collect(),
+        cache,
+    })
+}
+
 #[async_trait::async_trait]
 impl ScannerIOCache for SetDisks {
     #[tracing::instrument(skip(self, budget, scan_plan, updates))]
@@ -27,8 +114,9 @@ impl ScannerIOCache for SetDisks {
         scan_mode: HealScanMode,
     ) -> Result<()> {
         let ScannerBucketScanPlan {
-            buckets,
+            mut buckets,
             all_buckets,
+            scope,
             digest: scan_plan_digest,
             leader_epoch,
             tier_registry_generation,
@@ -63,26 +151,57 @@ impl ScannerIOCache for SetDisks {
                 "Scanner old data usage cache load failed; rebuilding from bucket caches"
             );
         }
+        let scoped_scan = prepare_scoped_set_scan(
+            &old_cache,
+            &buckets,
+            &all_buckets,
+            &scope,
+            ScannerSetCacheGeneration {
+                want_cycle,
+                leader_epoch,
+                tier_registry_generation,
+                source,
+                scan_plan_digest,
+            },
+        );
+        let mut scoped_cache = scoped_scan.map(|prepared| {
+            buckets = prepared.buckets;
+            prepared.cache
+        });
         if buckets.is_empty() {
             let now = SystemTime::now();
-            let mut cache = DataUsageCache {
-                info: DataUsageCacheInfo {
-                    name: DATA_USAGE_ROOT.to_string(),
-                    next_cycle: want_cycle,
-                    last_update: Some(now),
-                    leader_epoch,
-                    tier_registry_generation: Some(tier_registry_generation),
-                    source: Some(source),
-                    snapshot_complete: true,
-                    scan_plan_digest: Some(scan_plan_digest),
-                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-                    ..Default::default()
-                },
-                cache: HashMap::new(),
+            let mut cache = match scoped_cache.take() {
+                Some(cache) => cache,
+                None => {
+                    let mut cache = DataUsageCache {
+                        info: DataUsageCacheInfo {
+                            name: DATA_USAGE_ROOT.to_string(),
+                            next_cycle: want_cycle,
+                            leader_epoch,
+                            tier_registry_generation: Some(tier_registry_generation),
+                            source: Some(source),
+                            scan_plan_digest: Some(scan_plan_digest),
+                            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                            ..Default::default()
+                        },
+                        cache: HashMap::new(),
+                    };
+                    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+                    for bucket in all_buckets.iter() {
+                        cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+                    }
+                    cache
+                }
             };
-            cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
-            for bucket in all_buckets.iter() {
-                cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+            cache.info.last_update = Some(now);
+            cache.info.snapshot_complete = true;
+            cache.info.lkg_snapshot_complete = false;
+            cache.info.lkg_next_cycle = None;
+            cache.info.lkg_last_update = None;
+            cache.info.lkg_leader_epoch = None;
+            cache.info.lkg_scan_plan_digest = None;
+            if cache.find(DATA_USAGE_ROOT).is_none() {
+                cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
             }
             reset_disk_bucket_scan_gauges(&pool_label, &set_label);
             return persist_and_publish_cache_snapshot(
@@ -269,92 +388,102 @@ impl ScannerIOCache for SetDisks {
         record_disk_bucket_scans_active(0, &pool_label, &set_label);
         let _reset_disk_bucket_scan_gauges = DiskBucketScanGaugeReset::new(pool_label.clone(), set_label.clone());
 
-        // Fence a stale set aggregate before copying entries into per-bucket work caches.
-        if old_cache.info.next_cycle <= want_cycle
-            && old_cache.info.leader_epoch <= leader_epoch
-            && old_cache.info.tier_registry_generation != Some(tier_registry_generation)
-        {
-            old_cache.info.scan_plan_digest = None;
-        }
-        let old_lkg = old_cache.info.snapshot_complete.then_some({
-            (
-                old_cache.info.next_cycle,
-                old_cache.info.last_update,
-                old_cache.info.leader_epoch,
-                old_cache.info.scan_plan_digest,
-            )
-        });
-        let prepare_outcome = match old_cache.prepare_for_scan(
-            DATA_USAGE_ROOT,
-            want_cycle,
-            leader_epoch,
-            source,
-            scan_plan_digest,
-            require_cache_source,
-        ) {
-            DataUsageCachePrepareOutcome::RejectedNewerCycle => {
-                cache_cycle_floor.fetch_max(old_cache.info.next_cycle, Ordering::AcqRel);
-                warn!(
-                    target: "rustfs::scanner::io",
-                    event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_IO,
-                    pool = self.pool_index,
-                    set = self.set_index,
-                    cache_name = DATA_USAGE_CACHE_NAME,
-                    requested_cycle = want_cycle,
-                    cached_cycle = old_cache.info.next_cycle,
-                    state = "stale_cycle_rejected",
-                    "Scanner rejected a set cache cycle regression"
-                );
-                return Ok(());
+        let mut cache = if let Some(cache) = scoped_cache.take() {
+            cache
+        } else {
+            // Fence a stale set aggregate before copying entries into per-bucket work caches.
+            if old_cache.info.next_cycle <= want_cycle
+                && old_cache.info.leader_epoch <= leader_epoch
+                && old_cache.info.tier_registry_generation != Some(tier_registry_generation)
+            {
+                old_cache.info.scan_plan_digest = None;
             }
-            DataUsageCachePrepareOutcome::RejectedNewerLeader => {
-                warn!(
-                    target: "rustfs::scanner::io",
-                    event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_IO,
-                    pool = self.pool_index,
-                    set = self.set_index,
-                    cache_name = DATA_USAGE_CACHE_NAME,
-                    requested_epoch = leader_epoch,
-                    cached_epoch = old_cache.info.leader_epoch,
-                    state = "stale_leader_rejected",
-                    "Scanner rejected work from an older leader epoch"
-                );
-                return Ok(());
-            }
-            outcome => outcome,
-        };
-        if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused)
-            && let Some((cycle, last_update, epoch, digest)) = old_lkg
-        {
-            old_cache.info.lkg_snapshot_complete = true;
-            old_cache.info.lkg_next_cycle = Some(cycle);
-            old_cache.info.lkg_last_update = last_update;
-            old_cache.info.lkg_leader_epoch = Some(epoch);
-            old_cache.info.lkg_scan_plan_digest = digest;
-        }
-
-        let mut cache = DataUsageCache {
-            info: DataUsageCacheInfo {
-                name: DATA_USAGE_ROOT.to_string(),
-                next_cycle: want_cycle,
+            let old_lkg = old_cache.info.snapshot_complete.then_some({
+                (
+                    old_cache.info.next_cycle,
+                    old_cache.info.last_update,
+                    old_cache.info.leader_epoch,
+                    old_cache.info.scan_plan_digest,
+                )
+            });
+            let prepare_outcome = match old_cache.prepare_for_scan(
+                DATA_USAGE_ROOT,
+                want_cycle,
                 leader_epoch,
-                tier_registry_generation: Some(tier_registry_generation),
-                source: Some(source),
-                snapshot_complete: false,
-                scan_plan_digest: Some(scan_plan_digest),
-                cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-                ..Default::default()
-            },
-            cache: HashMap::new(),
+                source,
+                scan_plan_digest,
+                require_cache_source,
+            ) {
+                DataUsageCachePrepareOutcome::RejectedNewerCycle => {
+                    cache_cycle_floor.fetch_max(old_cache.info.next_cycle, Ordering::AcqRel);
+                    warn!(
+                        target: "rustfs::scanner::io",
+                        event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_IO,
+                        pool = self.pool_index,
+                        set = self.set_index,
+                        cache_name = DATA_USAGE_CACHE_NAME,
+                        requested_cycle = want_cycle,
+                        cached_cycle = old_cache.info.next_cycle,
+                        state = "stale_cycle_rejected",
+                        "Scanner rejected a set cache cycle regression"
+                    );
+                    return Ok(());
+                }
+                DataUsageCachePrepareOutcome::RejectedNewerLeader => {
+                    warn!(
+                        target: "rustfs::scanner::io",
+                        event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_IO,
+                        pool = self.pool_index,
+                        set = self.set_index,
+                        cache_name = DATA_USAGE_CACHE_NAME,
+                        requested_epoch = leader_epoch,
+                        cached_epoch = old_cache.info.leader_epoch,
+                        state = "stale_leader_rejected",
+                        "Scanner rejected work from an older leader epoch"
+                    );
+                    return Ok(());
+                }
+                outcome => outcome,
+            };
+            if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused)
+                && let Some((cycle, last_update, epoch, digest)) = old_lkg
+            {
+                old_cache.info.lkg_snapshot_complete = true;
+                old_cache.info.lkg_next_cycle = Some(cycle);
+                old_cache.info.lkg_last_update = last_update;
+                old_cache.info.lkg_leader_epoch = Some(epoch);
+                old_cache.info.lkg_scan_plan_digest = digest;
+            }
+
+            let mut cache = DataUsageCache {
+                info: DataUsageCacheInfo {
+                    name: DATA_USAGE_ROOT.to_string(),
+                    next_cycle: want_cycle,
+                    leader_epoch,
+                    tier_registry_generation: Some(tier_registry_generation),
+                    source: Some(source),
+                    snapshot_complete: false,
+                    scan_plan_digest: Some(scan_plan_digest),
+                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                    lkg_snapshot_complete: old_cache.info.lkg_snapshot_complete,
+                    lkg_next_cycle: old_cache.info.lkg_next_cycle,
+                    lkg_last_update: old_cache.info.lkg_last_update,
+                    lkg_leader_epoch: old_cache.info.lkg_leader_epoch,
+                    lkg_scan_plan_digest: old_cache.info.lkg_scan_plan_digest,
+                    ..Default::default()
+                },
+                cache: HashMap::new(),
+            };
+            cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+            for bucket in all_buckets.iter() {
+                cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+            }
+            cache
         };
-        cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
-        for bucket in all_buckets.iter() {
-            cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
-        }
 
         let (bucket_tx, bucket_rx) = mpsc::channel::<BucketInfo>(buckets.len());
 
@@ -1257,11 +1386,6 @@ impl ScannerIOCache for SetDisks {
             incomplete_scope.info.snapshot_complete = false;
             incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest);
             incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
-            incomplete_scope.info.lkg_snapshot_complete = old_cache.info.lkg_snapshot_complete;
-            incomplete_scope.info.lkg_next_cycle = old_cache.info.lkg_next_cycle;
-            incomplete_scope.info.lkg_last_update = old_cache.info.lkg_last_update;
-            incomplete_scope.info.lkg_leader_epoch = old_cache.info.lkg_leader_epoch;
-            incomplete_scope.info.lkg_scan_plan_digest = old_cache.info.lkg_scan_plan_digest;
             if let Err(e) = updates.send(incomplete_scope).await {
                 error!(
                     target: "rustfs::scanner::io",

crates/scanner/src/scanner_io/io_cycle.rs

@@ -63,6 +63,41 @@ pub(crate) async fn nsscanner_with_storage_status<S>(
 where
     S: ScannerStorage,
 {
+    let request = ScannerCycleRequest {
+        ctx,
+        budget,
+        updates,
+        want_cycle,
+        leader_epoch,
+        scan_mode,
+        scan_scope: ScannerBucketScanScope::default(),
+    };
+    nsscanner_with_storage_status_scoped(store, request).await
+}
+
+pub(crate) struct ScannerCycleRequest {
+    pub(crate) ctx: CancellationToken,
+    pub(crate) budget: Arc<ScannerCycleBudget>,
+    pub(crate) updates: mpsc::Sender<DataUsageInfo>,
+    pub(crate) want_cycle: u64,
+    pub(crate) leader_epoch: u64,
+    pub(crate) scan_mode: HealScanMode,
+    pub(crate) scan_scope: ScannerBucketScanScope,
+}
+
+pub(crate) async fn nsscanner_with_storage_status_scoped<S>(store: &S, request: ScannerCycleRequest) -> Result<ScannerCycleResult>
+where
+    S: ScannerStorage,
+{
+    let ScannerCycleRequest {
+        ctx,
+        budget,
+        updates,
+        want_cycle,
+        leader_epoch,
+        scan_mode,
+        scan_scope,
+    } = request;
     let child_token = ctx.child_token();
     let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
 
@@ -280,6 +315,7 @@ where
         let scan_plan = ScannerBucketScanPlan {
             buckets: set_buckets,
             all_buckets: Arc::clone(&all_buckets),
+            scope: scan_scope.clone(),
             digest: scan_plan_digest,
             leader_epoch,
             tier_registry_generation,

crates/scanner/src/scanner_io/tests.rs

@@ -765,6 +765,155 @@ fn bucket_usage_scan_order_prioritizes_dirty_buckets() {
     assert_eq!(names, vec!["dirty", "missing", "cached"]);
 }
 
+fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsageScanPlanDigest) -> DataUsageCache {
+    let mut cache = DataUsageCache {
+        info: DataUsageCacheInfo {
+            name: DATA_USAGE_ROOT.to_string(),
+            next_cycle: 7,
+            last_update: Some(SystemTime::now()),
+            leader_epoch: 11,
+            source: Some(DataUsageCacheSource::new(1, 2)),
+            snapshot_complete: true,
+            scan_plan_digest: Some(scan_plan_digest),
+            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+            tier_registry_generation: Some(13),
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+    for (bucket, size) in buckets {
+        cache.replace(
+            bucket,
+            DATA_USAGE_ROOT,
+            DataUsageEntry {
+                size: *size,
+                objects: 1,
+                ..Default::default()
+            },
+        );
+    }
+    cache
+}
+
+#[test]
+fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
+    let baseline_digest = DataUsageScanPlanDigest([1; 32]);
+    let current_digest = DataUsageScanPlanDigest([2; 32]);
+    let mut old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20), ("deleted", 30)], baseline_digest);
+    old_cache.replace(
+        "stable/prefix",
+        "stable",
+        DataUsageEntry {
+            size: 5,
+            objects: 1,
+            ..Default::default()
+        },
+    );
+    let all_buckets = vec![bucket_info("stable"), bucket_info("dirty")];
+    let selected_buckets = Arc::new(HashSet::from(["dirty".to_string(), "deleted".to_string()]));
+
+    let prepared = prepare_scoped_set_scan(
+        &old_cache,
+        &all_buckets,
+        &all_buckets,
+        &ScannerBucketScanScope {
+            selected_buckets: Some(selected_buckets),
+            baseline_scan_plan_digest: Some(baseline_digest),
+        },
+        ScannerSetCacheGeneration {
+            want_cycle: 8,
+            leader_epoch: 11,
+            tier_registry_generation: 13,
+            source: DataUsageCacheSource::new(1, 2),
+            scan_plan_digest: current_digest,
+        },
+    )
+    .expect("complete matching set cache should support a scoped scan");
+
+    assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(), ["dirty"]);
+    let stable = prepared
+        .cache
+        .checked_flatten("stable")
+        .expect("unselected bucket subtree should be retained");
+    assert_eq!((stable.size, stable.objects), (15, 2));
+    assert_eq!(prepared.cache.find("dirty").map(|entry| (entry.size, entry.objects)), Some((0, 0)));
+    assert!(prepared.cache.find("deleted").is_none());
+    assert_eq!(prepared.cache.info.scan_plan_digest, Some(current_digest));
+    assert_eq!(prepared.cache.info.next_cycle, 8);
+    assert!(!prepared.cache.info.snapshot_complete);
+    assert!(prepared.cache.info.lkg_snapshot_complete);
+    assert_eq!(prepared.cache.info.lkg_next_cycle, Some(7));
+    assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest));
+}
+
+#[test]
+fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
+    let baseline_digest = DataUsageScanPlanDigest([3; 32]);
+    let old_cache = complete_set_usage_cache(&[("stable", 10)], baseline_digest);
+    let all_buckets = vec![bucket_info("stable"), bucket_info("new")];
+
+    assert!(
+        prepare_scoped_set_scan(
+            &old_cache,
+            &all_buckets,
+            &all_buckets,
+            &ScannerBucketScanScope {
+                selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
+                baseline_scan_plan_digest: Some(baseline_digest),
+            },
+            ScannerSetCacheGeneration {
+                want_cycle: 8,
+                leader_epoch: 11,
+                tier_registry_generation: 13,
+                source: DataUsageCacheSource::new(1, 2),
+                scan_plan_digest: DataUsageScanPlanDigest([4; 32]),
+            },
+        )
+        .is_none()
+    );
+}
+
+#[test]
+fn scoped_set_scan_requires_an_exact_complete_baseline() {
+    let baseline_digest = DataUsageScanPlanDigest([5; 32]);
+    let all_buckets = vec![bucket_info("dirty")];
+    let scope = ScannerBucketScanScope {
+        selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
+        baseline_scan_plan_digest: Some(baseline_digest),
+    };
+    let generation = ScannerSetCacheGeneration {
+        want_cycle: 8,
+        leader_epoch: 11,
+        tier_registry_generation: 13,
+        source: DataUsageCacheSource::new(1, 2),
+        scan_plan_digest: DataUsageScanPlanDigest([6; 32]),
+    };
+
+    let mut incomplete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    incomplete.info.snapshot_complete = false;
+    assert!(prepare_scoped_set_scan(&incomplete, &all_buckets, &all_buckets, &scope, generation).is_none());
+
+    let mut not_durable = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    not_durable.info.last_update = None;
+    assert!(prepare_scoped_set_scan(&not_durable, &all_buckets, &all_buckets, &scope, generation).is_none());
+
+    let mut wrong_digest = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    wrong_digest.info.scan_plan_digest = Some(DataUsageScanPlanDigest([7; 32]));
+    assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation).is_none());
+
+    let empty_scope = ScannerBucketScanScope {
+        selected_buckets: Some(Arc::new(HashSet::new())),
+        baseline_scan_plan_digest: Some(baseline_digest),
+    };
+    let complete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation).is_none());
+
+    let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    future_cache.info.next_cycle = generation.want_cycle.saturating_add(1);
+    assert!(prepare_scoped_set_scan(&future_cache, &all_buckets, &all_buckets, &scope, generation).is_none());
+}
+
 #[test]
 fn record_set_scan_failure_preserves_first_error() {
     let mut first = None;

docs/architecture/ilm-tiering-persistence-contracts.md

@@ -1,7 +1,7 @@
 # ILM And Tiering Persistence Contracts
 
 **Use this when:** changing an ILM transition, tier configuration mutation, manual transition job, tier-delete recovery path, pool decommission, or any code that can create, transfer, or destroy ownership of a remote-tier object.
-**Source of truth:** `TransitionTransaction` and `process_transition_transaction_record` in `crates/ecstore/src/bucket/lifecycle/transition_transaction.rs`; `TierMutationIntent` and its conditional store helpers in `crates/ecstore/src/services/tier/tier_mutation_intent.rs`; `TierConfigMgr::update_candidate_with_config_lock` and mutation recovery in `crates/ecstore/src/services/tier/tier.rs`; `handle_tier_mutation_peer_request` in `crates/ecstore/src/services/tier/tier_mutation_peer.rs`; the record encoders and CAS helpers in `crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs`; manual-job execution/recovery and `cleanup_free_version_exact` in `crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs`; `process_tier_delete_journal_entry` and manifest recovery in `crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs`; the free-version scan/re-enqueue path `recover_tier_free_versions` in `crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs`; `DURABLE_ILM_NAMESPACES` and `validate_durable_ilm_record` in `crates/ecstore/src/bucket/lifecycle/durable_namespace.rs`; and `record_durable_ilm_decommission_progress`, receipt verification, and receipt cleanup in `crates/ecstore/src/core/pools.rs`.
+**Source of truth:** `TransitionTransaction` and `process_transition_transaction_record` in `crates/ecstore/src/bucket/lifecycle/transition_transaction.rs`; `TierMutationIntent` and its conditional store helpers in `crates/ecstore/src/services/tier/tier_mutation_intent.rs`; the dormant validation-probe record and conditional primitives in `crates/ecstore/src/services/tier/tier_probe_intent.rs`; `TierConfigMgr::update_candidate_with_config_lock` and mutation recovery in `crates/ecstore/src/services/tier/tier.rs`; `handle_tier_mutation_peer_request` in `crates/ecstore/src/services/tier/tier_mutation_peer.rs`; the record encoders and CAS helpers in `crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs`; manual-job execution/recovery and `cleanup_free_version_exact` in `crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs`; `process_tier_delete_journal_entry` and manifest recovery in `crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs`; the free-version scan/re-enqueue path `recover_tier_free_versions` in `crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs`; `DURABLE_ILM_NAMESPACES` and `validate_durable_ilm_record` in `crates/ecstore/src/bucket/lifecycle/durable_namespace.rs`; and `record_durable_ilm_decommission_progress`, receipt verification, and receipt cleanup in `crates/ecstore/src/core/pools.rs`.
 
 This document separates three kinds of statement:
 
@@ -44,6 +44,7 @@ All keys below are objects in the internal metadata bucket. The table gives the
 | Transition transaction | `rustfs-transition-transaction-v1` | `ilm/transition-transactions/records/<aa>/<bb>/<transaction-id>.json` | The transition attempt creates it; transition commit/recovery cleans it | Immutable/fence identity: deployment, transaction, fixed `owner_epoch`, write, source identity, tier/backend fingerprint, canonical remote object, deadline. Mutable: state, remote version, revision. `TransitionCleanupProof` is only a transient admission input to `mark_cleanup_pending`; it is not persisted in the record | Maximum-parity config write. Current create/update/delete calls do not use ETag preconditions |
 | Tier mutation peer intent | `rustfs-tier-mutation-intent-v1` | `tier/mutation-intents/records/<aa>/<bb>/<mutation-id>.json` | The receiving peer creates and converges it; the mutation recovery path cleans it | Immutable: mutation ID/kind, old config ETag, candidate digest, sorted affected target identities, expiry. Mutable: revision, state, committed config ETag | Create with `If-None-Match: *`; transition/delete with ETag `If-Match`; maximum parity |
 | Tier mutation coordinator intent | `rustfs-tier-mutation-intent-v1` | `tier/mutation-intents/coordinators/<aa>/<bb>/<mutation-id>.json` | The initiating node creates it; coordinator recovery cleans it after peer convergence | Same mutation identity and mutable fields as the peer record | Same conditional-write contract as the peer intent |
+| Tier validation probe intent | Dormant `rustfs-tier-probe-intent-v1`; no writer or recovery is enabled | `ilm/tier-probe-intents/records/<aa>/<bb>/<probe-id>.json` | No current runtime owner because no path creates the record; v1 permits only the immutable creator as owner | Immutable probe, operation-generation, destination, random remote object, creator identity, and v1 owner fence. Mutable: revision, state, and monotonic remote-version proof | Conditional create/CAS/delete primitives exist but are not called by Add/Edit/Verify or recovery |
 | Manual job | `rustfs-manual-transition-job-v1` | `ilm/manual-transition/jobs/<aa>/<bb>/<job-id>.json` | The admin run creates it; the active owner or recovery lease advances it. There is no current record GC owner | Immutable: job ID, bucket-level scope, options, creation time. Mutable: owner/lease, state, cancel bit, cursor, progress/report, queue snapshot, timestamps/error | Initial UUID-key write uses maximum parity without create-only precondition; later updates use ETag CAS |
 | Manual scope admission | `rustfs-manual-transition-job-v1` | `ilm/manual-transition/scopes/<aa>/<bb>/<scope-digest>.json` | Job admission creates/renews it; the job owner removes it after terminalization | Immutable bucket/run-vs-dry-run scope; mutable job/lease ownership and expiry | Create-only, renew/delete by ETag CAS |
 | Manual task | `rustfs-manual-transition-task-v1` | `ilm/manual-transition/tasks/<job shards>/<job-id>/<task-key>.json` | The scanner persists it before queue admission; no current GC owner | Immutable job plus exact bucket/object/version/tier work identity | Append-only create with `If-None-Match: *` and maximum parity |
@@ -55,7 +56,7 @@ All keys below are objects in the internal metadata bucket. The table gives the
 | Decommission durable-namespace receipt | `v2` | `decommission/ilm-receipts/<run-token>/<source-path>/<id-kind>/<id>.json` | The decommission coordinator writes target/source proof and is the only cleanup owner for that run | Source path, namespace and record identity, monotonic checkpoint, optional terminal checkpoint, optional v6 topology generation | Create-only then ETag CAS merge; checksum envelope; maximum parity |
 | Decommission expected-receipt manifest | `v1` | `decommission/ilm-manifests/<run-token>.json` | The source-pool decommission coordinator creates and cleans it | Run token plus exact sorted receipt-path count/digest | Create-only, exact readback, and verification before pool removal |
 
-`durable_namespace.rs` registers exactly the two tier-journal namespaces, the dispatch-record namespace shared by single manifests, chunk children, and chunk parents, the transaction namespace, and four manual-job namespaces. A path beginning with `ilm/` that is not in that registry is an error during decommission rather than an ignorable object.
+`durable_namespace.rs` registers exactly the two tier-journal namespaces, the dispatch-record namespace shared by single manifests, chunk children, and chunk parents, the transition-transaction namespace, the dormant tier-validation-probe namespace, and four manual-job namespaces. A path beginning with `ilm/` that is not in that registry is an error during decommission rather than an ignorable object.
 
 ## Durable fences and write primitives
 
@@ -187,6 +188,35 @@ Intent transitions retry an ETag race at most three times before returning a ret
 - **Open:** the exact split between backend validation, peer fanout, reference scan, config CAS, and publication; expiry must never replace config-generation proof.
 - **Open:** a dedicated operator reconcile/status surface and bounded retention for irreconcilable coordinator/peer records.
 
+## Tier validation probe intent
+
+### Dormant current contract
+
+`TierProbeIntent` defines a strict, checksum-protected `rustfs-tier-probe-intent-v1` envelope and the canonical key `ilm/tier-probe-intents/records/<aa>/<bb>/<probe-id>.json`. The same non-nil probe UUID also determines the remote object name `rustfs-tier-probe-<probe-id>`. The path parser rejects uppercase UUID aliases, wrong shards, extra components, and path/payload/object-name disagreement. The durable namespace registry validates this record so pool decommission cannot silently treat it as an ordinary object.
+
+The format binds Add and Edit to the same durable mutation identity tuple `(mutation_id, old_config_etag, candidate_digest)`. The old config ETag is required even for Add because it identifies the complete persisted tier configuration generation, not whether the named destination tier already exists. Verify instead requires the current persisted config ETag and credential-independent backend identity. These are mutually exclusive tagged variants. Every record also requires the tier name, matching destination identity, immutable creator identity/epoch, positive creation time, and an owner fence with nonempty owner, non-nil epoch, and later `not_after` timestamp. In v1 that owner identity must remain exactly equal to the immutable creator identity. It never persists the process-local driver revision or credential-bearing driver fingerprint.
+
+The dormant state graph is:
+
+```text
+UploadOutcomeUnknown -> Uploaded -> CleanupPending -> Completed
+                     \-> CleanupPending -> Completed
+                     \-> AbortedNoRemote
+```
+
+`UploadOutcomeUnknown` and `AbortedNoRemote` carry no remote version. `Uploaded`, `CleanupPending`, and `Completed` carry either explicit unversioned semantics or one exact nonempty opaque version. Once known, that remote version cannot change. Revision advances by one for each edge: `UploadOutcomeUnknown` is revision 1, `Uploaded` and `AbortedNoRemote` are revision 2, `CleanupPending` is revision 2 or 3, and `Completed` is revision 3 or 4. The strict decoder rejects any other state/revision pairing. The direct `UploadOutcomeUnknown -> CleanupPending` edge is reserved for a future authoritative provider probe that discovers the exact cleanup candidate after the original PUT response was lost.
+
+Create-only, ETag CAS, exact-ETag delete, and read-with-ETag primitives exist for the record, plus a crate-level read-only inspection result that always reports both writer and destructive recovery as disabled. No Add, Edit, Verify, startup loop, periodic loop, admin HTTP route, or remote backend operation currently calls these mutation primitives. Consequently this version creates no records and authorizes no remote PUT or DELETE.
+
+### Activation requirements
+
+- Add/Edit must create the mutation identity before validation and reread the same `(mutation_id, old_config_etag, candidate_digest)` before every probe-intent successor. Verify must reread the same config ETag, tier, and backend identity. A credential rotation may supply usable current credentials only when the persisted destination identity remains exact; it may not weaken operation-generation checks.
+- Before enabling any writer, every required node must advertise a probe-intent-specific read/retain/recovery capability. This protocol does not reuse the legacy transition-state reconciliation capability or its token. Unknown or older nodes keep validation in the current process-local mode and no durable v1 record is written.
+- The v1 owner fence is immutable and must equal the creator identity. Any future takeover requires a new schema with explicit takeover proof, plus an approved lease duration, clock-skew allowance, durable owner/epoch CAS, and revalidation order. Expiry alone never permits remote DELETE.
+- Remote-version discovery and deletion must use the provider-bound, destination-bound implementation and bounded request API approved for that target. Unknown, multiple, changing, unsupported, or unavailable results retain the record.
+- A successful or response-lost state write must strongly reread the exact record. Remote DELETE requires a current operation-generation proof, exact destination, current credentials for that same destination, a valid fleet fence, durable takeover, and the same known remote version immediately before and after the call.
+- Terminal retention, bounded scanning, metrics, and any HTTP inspect/reconcile route remain unapproved. Raw age is never cleanup evidence, and this dormant core API must not be presented as an operator endpoint.
+
 ## Manual transition job, task, result, and checkpoint
 
 ### Current contract

docs/operations/tier-ilm-debugging.md

@@ -16,6 +16,7 @@
 | Persisted free-version scan and re-enqueue after local-first expiry | `crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs` |
 | Fenced free-version remote delete, local-marker cleanup, and rescan | `crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs` (`cleanup_free_version_exact`) |
 | Durable manual transition job/task/result records | `crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs` |
+| Dormant tier validation probe intent format and read-only core inspection | `crates/ecstore/src/services/tier/tier_probe_intent.rs` |
 | Manual run/status/cancel and transition-transaction reconcile admin routes | `rustfs/src/admin/handlers/ilm_transition.rs` |
 | `ObjectInfo` / `TransitionedObject` types | `crates/ecstore/src/object_api/types.rs` |
 | `FileMeta` / `FileInfo` / version metadata | `crates/filemeta/src/` |
@@ -119,6 +120,12 @@ rc admin ilm transition run local/mybucket --prefix logs/ --tier cold --dry-run
 rc admin ilm transition run local/mybucket --prefix logs/ --tier cold --max-objects 1000 --max-duration-seconds 30
 ```
 
+## Validation probe crash recovery status
+
+Tier Add, Edit, and Verify currently validate a destination with a unique `rustfs-tier-probe-<uuid>` object and perform bounded compensation while the process remains alive. The `rustfs-tier-probe-intent-v1` decoder, canonical durable namespace, conditional storage primitives, state machine, and crate-level inspection type are present only as a dormant foundation. No validation path writes this record, no startup or periodic recovery scans it, and no admin HTTP route exposes it. V1 requires the owner to remain exactly equal to the immutable creator; takeover would require a new schema with explicit proof. Both durable writing and destructive recovery remain disabled until the fleet capability, operation-generation revalidation, provider timeout, retention, and operator contracts are approved.
+
+Do not search the internal metadata bucket for these records as evidence that validation is crash recoverable: a current server does not create them. If a process is killed after the remote probe PUT but before cleanup, inspect the destination provider manually and retain ambiguous candidates. Never delete an empty or guessed version, and do not hand-create a probe intent to authorize cleanup.
+
 Inspect the aggregate counters before widening scope. Full object-key lists are intentionally not returned. If `RUSTFS_RPC_SECRET` or other credentials were pasted into an issue, chat, log, or ticket while debugging tiering, rotate them on every node, restart the cluster with the new value, and redact the exposed copy before sharing more diagnostics.
 
 ## Reconcile an unknown transition upload

rustfs/src/version.rs

@@ -14,8 +14,6 @@
 
 use const_str::concat;
 use shadow_rs::shadow;
-use std::path::Path;
-use std::process::Command;
 
 shadow!(build);
 
@@ -47,10 +45,6 @@ pub const DISPLAY_VERSION: &str = {
 
 type VersionParseResult = Result<(u32, u32, u32, Option<String>), Box<dyn std::error::Error>>;
 
-fn build_version_override() -> Option<&'static str> {
-    BUILD_VERSION_OVERRIDE.filter(|version| !version.is_empty())
-}
-
 fn version_ref(version: &str) -> String {
     if version.starts_with("refs/tags/") || version.starts_with('@') {
         version.to_string()
@@ -61,91 +55,7 @@ fn version_ref(version: &str) -> String {
 
 #[allow(clippy::const_is_empty)]
 pub fn get_version() -> String {
-    if let Some(version) = build_version_override() {
-        return version_ref(version);
-    }
-
-    // Get the latest tag
-    if let Ok(latest_tag) = get_latest_tag() {
-        // Check if current commit is newer than the latest tag
-        if is_head_newer_than_tag(&latest_tag) {
-            // If current commit is newer, increment the version number
-            if let Ok(new_version) = increment_version(&latest_tag) {
-                return format!("refs/tags/{new_version}");
-            }
-        }
-
-        // If current commit is the latest tag, or version increment failed, return current tag
-        return format!("refs/tags/{latest_tag}");
-    }
-
-    // If no tag exists, use original logic
-    if !build::TAG.is_empty() {
-        format!("refs/tags/{}", build::TAG)
-    } else if !build::SHORT_COMMIT.is_empty() {
-        format!("@{}", build::SHORT_COMMIT)
-    } else {
-        format!("refs/tags/{}", build::PKG_VERSION)
-    }
-}
-
-/// Get the latest git tag
-fn get_latest_tag() -> Result<String, Box<dyn std::error::Error>> {
-    let output = Command::new("git").args(["describe", "--tags", "--abbrev=0"]).output()?;
-
-    if output.status.success() {
-        let tag = String::from_utf8(output.stdout)?;
-        Ok(tag.trim().to_string())
-    } else {
-        Err("Failed to get latest tag".into())
-    }
-}
-
-/// Check if current HEAD is newer than specified tag
-fn is_head_newer_than_tag(tag: &str) -> bool {
-    is_head_newer_than_tag_in(Path::new("."), tag)
-}
-
-fn is_head_newer_than_tag_in(repo: &Path, tag: &str) -> bool {
-    let head = Command::new("git").current_dir(repo).args(["rev-parse", "HEAD"]).output();
-    let tag_commit = Command::new("git")
-        .current_dir(repo)
-        .args(["rev-list", "-n", "1", tag])
-        .output();
-
-    let (Ok(head), Ok(tag_commit)) = (head, tag_commit) else {
-        return false;
-    };
-
-    if !head.status.success() || !tag_commit.status.success() || head.stdout == tag_commit.stdout {
-        return false;
-    }
-
-    let output = Command::new("git")
-        .current_dir(repo)
-        .args(["merge-base", "--is-ancestor", tag, "HEAD"])
-        .output();
-
-    match output {
-        Ok(result) => result.status.success(),
-        Err(_) => false,
-    }
-}
-
-/// Increment version number (increase patch version)
-fn increment_version(version: &str) -> Result<String, Box<dyn std::error::Error>> {
-    // Parse version number, e.g. "1.0.0-alpha.19" -> (1, 0, 0, Some("alpha.19"))
-    let (major, minor, patch, pre_release) = parse_version(version)?;
-
-    // If there's a pre-release identifier, increment the pre-release version number
-    if let Some(pre) = pre_release
-        && let Some(new_pre) = increment_pre_release(&pre)
-    {
-        return Ok(format!("{major}.{minor}.{patch}-{new_pre}"));
-    }
-
-    // Otherwise increment patch version number
-    Ok(format!("{major}.{minor}.{}", patch + 1))
+    version_ref(DISPLAY_VERSION)
 }
 
 /// Parse version number
@@ -166,28 +76,6 @@ pub fn parse_version(version: &str) -> VersionParseResult {
     Ok((major, minor, patch, pre_release))
 }
 
-/// Increment pre-release version number
-fn increment_pre_release(pre_release: &str) -> Option<String> {
-    // Handle pre-release versions like "alpha.19"
-    let parts: Vec<&str> = pre_release.split('.').collect();
-    if parts.len() == 2
-        && let Ok(num) = parts[1].parse::<u32>()
-    {
-        return Some(format!("{}.{}", parts[0], num + 1));
-    }
-
-    // Handle pre-release versions like "alpha19"
-    if let Some(pos) = pre_release.rfind(|c: char| c.is_alphabetic()) {
-        let prefix = &pre_release[..=pos];
-        let suffix = &pre_release[pos + 1..];
-        if let Ok(num) = suffix.parse::<u32>() {
-            return Some(format!("{prefix}{}", num + 1));
-        }
-    }
-
-    None
-}
-
 /// Clean version string - removes common prefixes
 pub fn clean_version(version: &str) -> String {
     version
@@ -284,34 +172,6 @@ mod tests {
     use super::*;
     use tracing::debug;
 
-    fn run_git(repo: &Path, args: &[&str]) {
-        let status = Command::new("git").current_dir(repo).args(args).status().unwrap();
-        assert!(status.success(), "git command failed: git {}", args.join(" "));
-    }
-
-    #[test]
-    fn test_is_head_newer_than_tag_requires_strict_descendant() {
-        let repo = tempfile::tempdir().unwrap();
-        run_git(repo.path(), &["init", "--quiet"]);
-        run_git(repo.path(), &["config", "user.name", "RustFS Tests"]);
-        run_git(repo.path(), &["config", "user.email", "rustfs@example.com"]);
-        run_git(repo.path(), &["commit", "--allow-empty", "--quiet", "-m", "tagged commit"]);
-        run_git(repo.path(), &["tag", "--annotate", "1.2.3", "--message", "1.2.3"]);
-
-        assert!(!is_head_newer_than_tag_in(repo.path(), "1.2.3"));
-
-        run_git(repo.path(), &["commit", "--allow-empty", "--quiet", "-m", "newer commit"]);
-
-        assert!(is_head_newer_than_tag_in(repo.path(), "1.2.3"));
-    }
-
-    #[test]
-    fn build_version_override_is_used_for_current_version_when_set() {
-        if let Some(version) = build_version_override() {
-            assert_eq!(get_version(), version_ref(version));
-        }
-    }
-
     #[test]
     fn version_ref_keeps_existing_ref_prefixes() {
         assert_eq!(version_ref("1.2.3"), "refs/tags/1.2.3");
@@ -319,6 +179,11 @@ mod tests {
         assert_eq!(version_ref("@abc123"), "@abc123");
     }
 
+    #[test]
+    fn get_version_uses_build_metadata() {
+        assert_eq!(get_version(), version_ref(DISPLAY_VERSION));
+    }
+
     #[test]
     fn test_parse_version() {
         // Test standard version parsing
@@ -336,27 +201,6 @@ mod tests {
         assert_eq!(pre_release, Some("alpha.19".to_string()));
     }
 
-    #[test]
-    fn test_increment_pre_release() {
-        // Test alpha.19 -> alpha.20
-        assert_eq!(increment_pre_release("alpha.19"), Some("alpha.20".to_string()));
-
-        // Test beta.5 -> beta.6
-        assert_eq!(increment_pre_release("beta.5"), Some("beta.6".to_string()));
-
-        // Test unparsable case
-        assert_eq!(increment_pre_release("unknown"), None);
-    }
-
-    #[test]
-    fn test_increment_version() {
-        // Test pre-release version increment
-        assert_eq!(increment_version("1.0.0-alpha.19").unwrap(), "1.0.0-alpha.20");
-
-        // Test standard version increment
-        assert_eq!(increment_version("1.0.0").unwrap(), "1.0.1");
-    }
-
     #[test]
     fn test_version_format() {
         // Test if version format starts with refs/tags/