Merge branch 'main' into houseme/fix/scanner-heal-v2-w02

Resolved scanner cache snapshot provenance conflicts after the main branch
added complete usage set-state metadata.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

.config/make/pre-commit.mak

@@ -3,10 +3,9 @@
 .NOTPARALLEL: pre-commit pre-pr dev-check
 
 .PHONY: setup-hooks
-setup-hooks: ## Install the configured pre-commit hooks
+setup-hooks: ## Set up git hooks
 	@echo "🔧 Setting up git hooks..."
-	pre-commit validate-config
-	pre-commit install
+	chmod +x .git/hooks/pre-commit
 	@echo "✅ Git hooks setup complete!"
 
 .PHONY: doc-paths-check

.github/pull_request_template.md

@@ -10,16 +10,16 @@ Use N/A when there is no related issue.
 
 ## Summary of Changes
 <!--
-Briefly explain what changed and why reviewers should accept it.
-Focus on behavior, compatibility, and review-relevant context.
+Describe the concrete problem and resulting behavior. For a behavior change, name the input or state that triggers it and the expected outcome. Explain any new dependency or abstraction that the change needs.
 -->
 
 ## Verification
 <!--
-List the commands or checks you ran, for example:
-- `make pre-commit`
+Give 1–3 concrete pieces of evidence for the changed behavior: the test or command, its observed result, and the regression it catches. For a bug fix, record a failing-before/passing-after check or explain why it was unavailable.
 
-Use N/A only when verification is not applicable.
+Identify the tested commit and any local changes. When testing a prebuilt binary or external service, include its source/version and artifact identity; a successful run against a different build is not evidence for this change.
+
+List relevant checks not run and the remaining risk. Use the validation tier in AGENTS.md; do not run broader checks solely to fill this section. For documentation-only changes, list the applicable documentation checks. Use N/A only when verification is not applicable.
 -->
 
 ## Impact

.pre-commit-config.yaml

@@ -3,9 +3,9 @@
 repos:
   - repo: local
     hooks:
-      - id: rustfs-fmt-check
-        name: Rust formatting
-        entry: cargo fmt --all --check
+      - id: rustfs-dev-check
+        name: rustfs dev-check
+        entry: make dev-check
         language: system
         types: [rust]
         pass_filenames: false

CONTRIBUTING.md

@@ -109,17 +109,24 @@ affected boundaries and risks. CI still runs its configured repository gates.
 
 ### 🔒 Git Pre-commit Hooks (optional)
 
-The optional hook uses the checked-in `.pre-commit-config.yaml`. Install [pre-commit](https://pre-commit.com/#installation), then run this from the checkout or a linked worktree:
+Git hooks are **not** versioned in this repository, so a fresh clone has no
+active pre-commit hook. If you add your own `.git/hooks/pre-commit` (a good
+choice is a one-liner that runs `make pre-commit`), you can mark it executable
+with:
 
 ```bash
 make setup-hooks
 ```
 
-The hook runs `cargo fmt --all --check` when staged files include Rust source. It does not compile the workspace or run tests. Fix formatting with `cargo fmt --all`, inspect and stage the result, then commit again.
+Or manually:
 
-`pre-commit install` resolves Git's hook directory for linked worktrees and preserves an existing hook in migration mode. If you use `core.hooksPath`, keep that hook manager and integrate `pre-commit run` there; the installer refuses to silently replace that configuration.
+```bash
+chmod +x .git/hooks/pre-commit
+```
 
-A local hook provides early formatting feedback. With or without it, follow the verification tiers in `AGENTS.md`, run relevant behavioral tests, and satisfy the CI merge gates. `make pre-commit` and `make dev-check` remain explicit broader commands.
+With or without a hook, follow the verification tiers in `AGENTS.md`. Run the
+applicable scoped checks, and reserve `make pre-pr` for broad cross-module
+changes whose impact cannot be bounded by those checks.
 
 ### 📝 Formatting Configuration
 
@@ -131,11 +138,31 @@ fn_call_width = 90
 single_line_let_else_max_width = 100
 ```
 
+### 🚫 Commit Prevention
+
+If you set up a pre-commit hook and your code doesn't meet the formatting requirements, the hook will:
+
+1. **Block the commit** and show clear error messages
+2. **Provide exact commands** to fix the issues
+3. **Guide you through** the resolution process
+
+Example output when formatting fails:
+
+```
+❌ Code formatting check failed!
+💡 Please run 'cargo fmt --all' to format your code before committing.
+
+🔧 Quick fix:
+   cargo fmt --all
+   git add .
+   git commit
+```
+
 ### 🔄 Development Workflow
 
 1. **Make your changes**
 2. **Format your code**: `make fmt` or `cargo fmt --all`
-3. **Select relevant checks** using the validation tier in `AGENTS.md`; use `make pre-commit` when its broader fast gate adds useful coverage
+3. **Run the fast gate**: `make pre-commit` (no clippy, no tests)
 4. **Commit your changes**: `git commit -m "your message"`
 5. **Complete the applicable multi-role adversarial review** for non-exempt changes (see `AGENTS.md`)
 6. **Run applicable scoped checks before opening/updating a PR**; consider
@@ -179,12 +206,11 @@ Configure your IDE to:
 #### Pre-commit hook not running?
 
 ```bash
-pre-commit validate-config
-pre-commit run --all-files
-# Inspect any configured hook manager; do not overwrite it.
-git config --get core.hooksPath
-# Install if no separate hook manager is configured.
-make setup-hooks
+# Check if hook is executable
+ls -la .git/hooks/pre-commit
+
+# Make it executable if needed
+chmod +x .git/hooks/pre-commit
 ```
 
 #### Formatting issues?

crates/scanner/src/data_usage_define.rs

@@ -774,6 +774,29 @@ impl DataUsageCache {
         (visited == expected_entries).then_some(entry)
     }
 
+    pub(crate) fn has_complete_root_inventory(&self, bucket_keys: &HashSet<String>) -> bool {
+        let Some(root) = self.find(DATA_USAGE_ROOT) else {
+            return false;
+        };
+        // Set roots only connect bucket entries. Scalar data at the root, an
+        // extra bucket, or an orphan must not disappear during bucket folding.
+        root.children.len() == bucket_keys.len()
+            && bucket_keys.iter().all(|key| root.children.contains(key))
+            && root.size == 0
+            && root.objects == 0
+            && root.versions == 0
+            && root.delete_markers == 0
+            && root.failed_objects == 0
+            && !root.compacted
+            && root.obj_sizes.is_empty()
+            && root.obj_versions.is_empty()
+            && root.replication_stats.is_none()
+            && root.all_tier_stats.is_none()
+            && root.unknown_tier_stats.is_none()
+            && root.tier_accounting_proof.is_none()
+            && self.checked_flatten_complete(DATA_USAGE_ROOT).is_some()
+    }
+
     fn checked_flatten_inner(&self, path: &str) -> Option<(DataUsageEntry, usize)> {
         let root_key = hash_path(path).key();
         let (root_key, root) = self.cache.get_key_value(&root_key)?;

crates/scanner/src/scanner/tests.rs

@@ -4758,6 +4758,28 @@ async fn usage_bootstrap_does_not_overwrite_concurrent_replacement() {
 #[serial]
 async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() {
     let (_temp_dir, store) = setup_scanner_cycle_store().await;
+    let quota_ledger_path = "config/quota-ledger/reserved-bucket.json";
+    let quota_ledger = serde_json::to_vec(&serde_json::json!({
+        "version": 1,
+        "bucket_incarnation": "00000000-0000-0000-0000-000000000001",
+        "quota_revision_unix_nanos": 1,
+        "accounted_usage": 100,
+        "reservations": {
+            "00000000-0000-0000-0000-000000000002": {
+                "object": "pending-object",
+                "old_size": 0,
+                "new_size": 64,
+                "created_at": 1,
+                "pool_index": 0,
+                "set_index": 0,
+                "commit_started": true
+            }
+        }
+    }))
+    .expect("quota ledger fixture should encode");
+    save_config(store.clone(), quota_ledger_path, quota_ledger.clone())
+        .await
+        .expect("independent quota reservations should persist");
     let cycle = CurrentCycle {
         current: 41,
         next: 42,
@@ -4816,6 +4838,14 @@ async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() {
     assert!(!data_usage_info_has_persisted_baseline_identity(&usage));
     assert_eq!(usage.scanner_epoch, Some(9));
 
+    assert_eq!(
+        read_config(store.clone(), quota_ledger_path)
+            .await
+            .expect("quota ledger must remain readable after scanner reset"),
+        quota_ledger,
+        "scanner reset must preserve incarnation and outstanding reserved bytes exactly"
+    );
+
     for path in [
         usage_backup_path.as_str(),
         LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(),

crates/scanner/src/scanner_io.rs

@@ -312,6 +312,30 @@ fn scanner_bucket_plan_digest(buckets: &[BucketInfo], activity_digest: [u8; 32])
     DataUsageScanPlanDigest(hasher.finalize().into())
 }
 
+fn scanner_bucket_inventory_is_complete(
+    all_buckets: &[BucketInfo],
+    buckets_by_source: &HashMap<DataUsageCacheSource, Vec<BucketInfo>>,
+) -> bool {
+    let inventory = all_buckets
+        .iter()
+        .map(|bucket| (bucket.name.as_str(), bucket.created))
+        .collect::<HashMap<_, _>>();
+    if inventory.len() != all_buckets.len() || inventory.keys().any(|name| name.is_empty() || *name == DATA_USAGE_ROOT) {
+        return false;
+    }
+    let mut covered = HashSet::with_capacity(inventory.len());
+    for buckets in buckets_by_source.values() {
+        let mut set_names = HashSet::with_capacity(buckets.len());
+        for bucket in buckets {
+            if !set_names.insert(bucket.name.as_str()) || inventory.get(bucket.name.as_str()) != Some(&bucket.created) {
+                return false;
+            }
+            covered.insert(bucket.name.as_str());
+        }
+    }
+    covered.len() == inventory.len()
+}
+
 fn scanner_bucket_cache_digest(
     scan_plan_digest: DataUsageScanPlanDigest,
     dirty_generation: Option<u64>,

crates/scanner/src/scanner_io/cache.rs

@@ -213,10 +213,85 @@ pub(super) fn cache_snapshot_is_current(
     )
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) struct ScannerSnapshotIdentity {
+    pub(super) cycle: u64,
+    pub(super) leader_epoch: u64,
+    pub(super) plan_digest: DataUsageScanPlanDigest,
+    pub(super) tier_registry_generation: Option<u64>,
+}
+
+pub(super) struct ScannerSnapshotScope<'a> {
+    pub(super) sources: &'a HashSet<DataUsageCacheSource>,
+    pub(super) buckets: &'a [String],
+    pub(super) identity: ScannerSnapshotIdentity,
+}
+
+#[derive(Debug, PartialEq, Eq, thiserror::Error)]
+pub(super) enum ScannerSnapshotValidationError {
+    #[error("scanner snapshot does not cover the expected complete sets")]
+    IncompleteSets,
+    #[error("scanner snapshot does not match the requested generation")]
+    GenerationMismatch,
+    #[error("scanner snapshot bucket inventory is invalid")]
+    InvalidInventory,
+    #[error("scanner snapshot root is incomplete or corrupt")]
+    InvalidRoot,
+}
+
+struct ValidatedScannerSnapshot<'a> {
+    results: &'a [DataUsageCache],
+    last_update: SystemTime,
+}
+
+impl<'a> ValidatedScannerSnapshot<'a> {
+    fn validate(
+        results: &'a [DataUsageCache],
+        scope: &ScannerSnapshotScope<'_>,
+    ) -> std::result::Result<Self, ScannerSnapshotValidationError> {
+        if !scanner_results_form_complete_snapshot(results, scope.sources) {
+            return Err(ScannerSnapshotValidationError::IncompleteSets);
+        }
+        let bucket_keys = scope
+            .buckets
+            .iter()
+            .map(|bucket| crate::hash_path(bucket).key())
+            .collect::<HashSet<_>>();
+        if bucket_keys.len() != scope.buckets.len()
+            || scope
+                .buckets
+                .iter()
+                .any(|bucket| bucket.is_empty() || bucket == DATA_USAGE_ROOT)
+        {
+            return Err(ScannerSnapshotValidationError::InvalidInventory);
+        }
+        for result in results {
+            if result.info.next_cycle != scope.identity.cycle
+                || result.info.leader_epoch != scope.identity.leader_epoch
+                || result.info.scan_plan_digest != Some(scope.identity.plan_digest)
+                || result.info.tier_registry_generation != scope.identity.tier_registry_generation
+            {
+                return Err(ScannerSnapshotValidationError::GenerationMismatch);
+            }
+            if result.info.name != DATA_USAGE_ROOT
+                || result.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
+                || !result.has_complete_root_inventory(&bucket_keys)
+            {
+                return Err(ScannerSnapshotValidationError::InvalidRoot);
+            }
+        }
+        let last_update = results
+            .iter()
+            .filter_map(|result| result.info.last_update)
+            .max()
+            .ok_or(ScannerSnapshotValidationError::IncompleteSets)?;
+        Ok(Self { results, last_update })
+    }
+}
+
 pub(super) fn completed_data_usage_info(
     results: &[DataUsageCache],
-    expected_sources: &HashSet<DataUsageCacheSource>,
-    all_buckets: &[String],
+    scope: &ScannerSnapshotScope<'_>,
     tier_registry_names: &[String],
     bucket_plan_complete: bool,
     budget_elapsed: bool,
@@ -229,26 +304,10 @@ pub(super) fn completed_data_usage_info(
     if !should_publish_completed_snapshot(completed_set_count, results.len(), budget_elapsed, cancelled) {
         return None;
     }
-    if !scanner_results_form_complete_snapshot(results, expected_sources) {
-        return None;
-    }
-
-    // A generation is comparable across nodes because it is derived from the
-    // frozen registry names. Cycle and leader fencing remain separate cache
-    // metadata. Legacy peers omit the generation; an all-legacy result remains
-    // readable, but mixing legacy and new (or two new generations) would make
-    // the per-tier accounting ambiguous.
-    let registry_generation = results.first()?.info.tier_registry_generation;
-    if results.iter().any(|result| match registry_generation {
-        Some(generation) => result.info.tier_registry_generation != Some(generation),
-        None => result.info.tier_registry_generation.is_some(),
-    }) {
-        return None;
-    }
-
-    if results.iter().any(|result| result.root().is_none()) {
-        return None;
-    }
+    let validated = ValidatedScannerSnapshot::validate(results, scope).ok()?;
+    let results = validated.results;
+    let all_buckets = scope.buckets;
+    let registry_generation = scope.identity.tier_registry_generation;
 
     let mut total = DataUsageEntry::default();
     let mut bucket_entries = HashMap::with_capacity(all_buckets.len());
@@ -273,7 +332,7 @@ pub(super) fn completed_data_usage_info(
         return None;
     }
 
-    let merged_last_update = results.iter().filter_map(|result| result.info.last_update).max()?;
+    let merged_last_update = validated.last_update;
     let buckets_usage = bucket_entries
         .iter()
         .map(|(bucket, entry)| Some((bucket.clone(), checked_bucket_usage_info(entry)?)))
@@ -300,8 +359,8 @@ pub(super) fn completed_data_usage_info(
     usage_snapshot_set_states.sort_by_key(|state| (state.pool_index, state.set_index));
     let data_usage_info = DataUsageInfo {
         last_update: Some(merged_last_update),
-        scanner_cycle: Some(results.first()?.info.next_cycle),
-        scanner_epoch: Some(results.first()?.info.leader_epoch),
+        scanner_cycle: Some(scope.identity.cycle),
+        scanner_epoch: Some(scope.identity.leader_epoch),
         objects_total_count: u64::try_from(total.objects).ok()?,
         versions_total_count: u64::try_from(total.versions).ok()?,
         delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,

crates/scanner/src/scanner_io/io_cache.rs

@@ -39,6 +39,12 @@ pub(super) fn prepare_scoped_set_scan(
     else {
         return None;
     };
+    // The existing cache does not bind each bucket to a durable incarnation.
+    // Listing creation times can come from volume metadata, so even Some(time)
+    // cannot prove that an unselected same-name bucket is the cached bucket.
+    if all_buckets.iter().any(|bucket| !selected_buckets.contains(&bucket.name)) {
+        return None;
+    }
     if selected_buckets.is_empty()
         || !old_cache.info.snapshot_complete
         || old_cache.info.last_update.is_none()
@@ -49,7 +55,7 @@ pub(super) fn prepare_scoped_set_scan(
         || 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()
+        || !old_cache.has_complete_root_inventory(&old_cache.find(DATA_USAGE_ROOT)?.children)
     {
         return None;
     }
@@ -74,21 +80,12 @@ pub(super) fn prepare_scoped_set_scan(
         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)?;
+        cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
     }
 
     Some(PreparedScopedSetScan {

crates/scanner/src/scanner_io/io_cycle.rs

@@ -260,6 +260,7 @@ where
         }
     }
     bucket_plan_complete &= buckets_by_source.keys().copied().collect::<HashSet<_>>() == *expected_sources;
+    bucket_plan_complete &= scanner_bucket_inventory_is_complete(&all_buckets, &buckets_by_source);
     let scan_plan_digest =
         scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before));
     let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list));
@@ -536,8 +537,16 @@ where
     let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::<Vec<_>>();
     let completed_usage = completed_data_usage_info(
         &results,
-        &expected_sources,
-        &all_bucket_names,
+        &ScannerSnapshotScope {
+            sources: &expected_sources,
+            buckets: &all_bucket_names,
+            identity: ScannerSnapshotIdentity {
+                cycle: want_cycle,
+                leader_epoch,
+                plan_digest: scan_plan_digest,
+                tier_registry_generation: Some(tier_registry_generation),
+            },
+        },
         &tier_registry.names,
         bucket_plan_complete,
         budget_elapsed,

crates/scanner/src/scanner_io/publish_gate_tests.rs

@@ -18,6 +18,32 @@ use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage, TierAccount
 
 const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);
 
+#[test]
+fn scanner_bucket_inventory_requires_exact_unique_set_union() {
+    let first = BucketInfo {
+        name: "first".to_string(),
+        ..Default::default()
+    };
+    let second = BucketInfo {
+        name: "second".to_string(),
+        ..Default::default()
+    };
+    let source = DataUsageCacheSource::new(0, 0);
+    let mut sets = HashMap::from([(source, vec![first.clone()])]);
+    assert!(scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
+    assert!(!scanner_bucket_inventory_is_complete(&[first.clone(), second.clone()], &sets));
+    assert!(!scanner_bucket_inventory_is_complete(&[], &sets));
+    assert!(!scanner_bucket_inventory_is_complete(&[first.clone(), first.clone()], &sets));
+    sets.insert(source, vec![first.clone(), first.clone()]);
+    assert!(!scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
+    sets.insert(source, vec![second]);
+    assert!(!scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
+    let mut recreated = first.clone();
+    recreated.created = Some(OffsetDateTime::UNIX_EPOCH);
+    sets.insert(source, vec![recreated]);
+    assert!(!scanner_bucket_inventory_is_complete(&[first], &sets));
+}
+
 #[test]
 fn should_publish_completed_snapshot_requires_full_clean_cycle() {
     assert!(should_publish_completed_snapshot(3, 3, false, false));
@@ -108,7 +134,133 @@ fn completed_data_usage_info_for_test(
     cancelled: bool,
 ) -> Option<(DataUsageInfo, SystemTime)> {
     let expected_sources = results.iter().filter_map(|result| result.info.source).collect::<HashSet<_>>();
-    completed_data_usage_info(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled)
+    completed_usage_for_scope(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled)
+}
+
+fn completed_usage_for_scope(
+    results: &[DataUsageCache],
+    expected_sources: &HashSet<DataUsageCacheSource>,
+    all_buckets: &[String],
+    tier_registry_names: &[String],
+    bucket_plan_complete: bool,
+    budget_elapsed: bool,
+    cancelled: bool,
+) -> Option<(DataUsageInfo, SystemTime)> {
+    let first = results.first()?;
+    completed_data_usage_info(
+        results,
+        &ScannerSnapshotScope {
+            sources: expected_sources,
+            buckets: all_buckets,
+            identity: ScannerSnapshotIdentity {
+                cycle: first.info.next_cycle,
+                leader_epoch: first.info.leader_epoch,
+                plan_digest: TEST_PLAN_DIGEST,
+                tier_registry_generation: first.info.tier_registry_generation,
+            },
+        },
+        tier_registry_names,
+        bucket_plan_complete,
+        budget_elapsed,
+        cancelled,
+    )
+}
+
+#[test]
+fn completed_data_usage_info_rejects_duplicate_bucket_inventory() {
+    let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
+    let buckets = vec!["bucket".to_string(), "bucket".to_string()];
+    assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none());
+}
+
+#[test]
+fn completed_data_usage_info_rejects_extra_or_detached_bucket_data() {
+    let buckets = vec!["bucket".to_string()];
+    let mut set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
+    set.replace(
+        "unlisted",
+        DATA_USAGE_ROOT,
+        DataUsageEntry {
+            objects: 1,
+            ..Default::default()
+        },
+    );
+    assert!(completed_data_usage_info_for_test(&[set.clone()], &buckets, false, false).is_none());
+    set.cache
+        .get_mut(DATA_USAGE_ROOT)
+        .expect("set root")
+        .children
+        .remove(&hash_path("unlisted").key());
+    assert!(
+        completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none(),
+        "orphaned data must not disappear from authoritative accounting"
+    );
+}
+
+#[test]
+fn completed_data_usage_info_rejects_disconnected_expected_bucket() {
+    let buckets = vec!["bucket".to_string()];
+    let mut set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
+    set.cache.get_mut(DATA_USAGE_ROOT).expect("set root").children.clear();
+    assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none());
+}
+
+#[test]
+fn completed_data_usage_info_rejects_root_scalar_data_and_unknown_key_format() {
+    let buckets = vec!["bucket".to_string()];
+    let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
+    let mut scalar_root = set.clone();
+    scalar_root.cache.get_mut(DATA_USAGE_ROOT).expect("set root").size = 10;
+    assert!(completed_data_usage_info_for_test(&[scalar_root], &buckets, false, false).is_none());
+    let mut future_format = set;
+    future_format.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT + 1;
+    assert!(completed_data_usage_info_for_test(&[future_format], &buckets, false, false).is_none());
+}
+
+#[test]
+fn completed_data_usage_info_binds_all_results_to_requested_identity() {
+    let buckets = vec!["bucket".to_string()];
+    let source = DataUsageCacheSource::new(0, 0);
+    let sources = HashSet::from([source]);
+    let set = completed_root_cache("bucket", 2, 10, source);
+    let identity = ScannerSnapshotIdentity {
+        cycle: 0,
+        leader_epoch: 0,
+        plan_digest: TEST_PLAN_DIGEST,
+        tier_registry_generation: None,
+    };
+    let results = [set];
+    for expected in [
+        ScannerSnapshotIdentity { cycle: 1, ..identity },
+        ScannerSnapshotIdentity {
+            leader_epoch: 1,
+            ..identity
+        },
+        ScannerSnapshotIdentity {
+            plan_digest: DataUsageScanPlanDigest([9; 32]),
+            ..identity
+        },
+        ScannerSnapshotIdentity {
+            tier_registry_generation: Some(1),
+            ..identity
+        },
+    ] {
+        let scope = ScannerSnapshotScope {
+            sources: &sources,
+            buckets: &buckets,
+            identity: expected,
+        };
+        assert!(completed_data_usage_info(&results, &scope, &[], true, false, false).is_none());
+    }
+    let scope = ScannerSnapshotScope {
+        sources: &sources,
+        buckets: &buckets,
+        identity,
+    };
+    let (usage, _) = completed_data_usage_info(&results, &scope, &[], true, false, false)
+        .expect("the requested complete scope remains publishable");
+    assert_eq!(usage.objects_total_count, 2);
+    assert!(usage.is_complete_bucket_usage_snapshot());
 }
 
 fn lkg_root_cache(bucket: &str, objects: usize, source: DataUsageCacheSource) -> DataUsageCache {
@@ -136,7 +288,7 @@ fn partial_usage_is_observational_not_authoritative_for_quota() {
     let expected = HashSet::from([current_source, stalled_source]);
 
     assert!(
-        completed_data_usage_info(&[current.clone(), stalled.clone()], &expected, &all_buckets, &[], true, false, false)
+        completed_usage_for_scope(&[current.clone(), stalled.clone()], &expected, &all_buckets, &[], true, false, false)
             .is_none()
     );
     let (observed, _) = observational_data_usage_info(&[current, stalled], &expected, &all_buckets, &[], TEST_PLAN_DIGEST, 8, 3)
@@ -509,7 +661,7 @@ fn completed_data_usage_info_accepts_unknown_only_with_current_registry_generati
 
     let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
     assert!(
-        completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some()
+        completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some()
     );
 }
 
@@ -547,7 +699,7 @@ fn completed_data_usage_info_rejects_non_registry_tier_in_current_generation() {
 
     let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
     assert!(
-        completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none()
+        completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none()
     );
 }
 
@@ -698,6 +850,7 @@ fn completed_data_usage_info_publishes_confirmed_empty_namespace() {
             source: Some(DataUsageCacheSource::new(0, 0)),
             snapshot_complete: true,
             scan_plan_digest: Some(TEST_PLAN_DIGEST),
+            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
             ..Default::default()
         },
         ..Default::default()
@@ -855,7 +1008,7 @@ fn completed_data_usage_info_requires_exact_topology_sources() {
     let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(1, 0)]);
 
     assert!(
-        completed_data_usage_info(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false)
+        completed_usage_for_scope(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false)
             .is_none()
     );
 }
@@ -866,7 +1019,7 @@ fn completed_data_usage_info_rejects_incomplete_bucket_plan() {
     let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
     let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
 
-    assert!(completed_data_usage_info(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none());
+    assert!(completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none());
 }
 
 #[test]

crates/scanner/src/scanner_io/tests.rs

@@ -797,6 +797,13 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa
     cache
 }
 
+fn bucket_info_with_created_time(name: &str) -> BucketInfo {
+    BucketInfo {
+        created: Some(time::OffsetDateTime::UNIX_EPOCH),
+        ..bucket_info(name)
+    }
+}
+
 fn complete_usage_baseline(
     source: DataUsageCacheSource,
     scan_plan_digest: DataUsageScanPlanDigest,
@@ -987,7 +994,7 @@ fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state()
 }
 
 #[test]
-fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
+fn scoped_set_scan_rebuilds_selected_buckets_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);
@@ -1000,8 +1007,11 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
             ..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 all_buckets = vec![
+        bucket_info_with_created_time("stable"),
+        bucket_info_with_created_time("dirty"),
+    ];
+    let selected_buckets = Arc::new(HashSet::from(["stable".to_string(), "dirty".to_string(), "deleted".to_string()]));
 
     let prepared = prepare_scoped_set_scan(
         &old_cache,
@@ -1021,12 +1031,16 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
     )
     .expect("complete matching set cache should support a scoped scan");
 
-    assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(), ["dirty"]);
+    assert_eq!(
+        prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(),
+        ["stable", "dirty"]
+    );
     let stable = prepared
         .cache
         .checked_flatten("stable")
-        .expect("unselected bucket subtree should be retained");
-    assert_eq!((stable.size, stable.objects), (15, 2));
+        .expect("selected bucket placeholder should exist");
+    assert_eq!((stable.size, stable.objects), (0, 0));
+    assert!(prepared.cache.find("stable/prefix").is_none());
     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));
@@ -1037,11 +1051,41 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
     assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest));
 }
 
+#[test]
+fn scoped_set_scan_rejects_unbound_bucket_incarnations() {
+    let baseline_digest = DataUsageScanPlanDigest([1; 32]);
+    let old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20)], baseline_digest);
+    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([2; 32]),
+    };
+    for created in [
+        None,
+        Some(OffsetDateTime::UNIX_EPOCH),
+        Some(OffsetDateTime::UNIX_EPOCH + time::Duration::days(1)),
+    ] {
+        let mut stable = bucket_info("stable");
+        stable.created = created;
+        let buckets = vec![stable, bucket_info_with_created_time("dirty")];
+        assert!(
+            prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation).is_none(),
+            "missing identity, volume timestamps and same-name recreation must all rebuild"
+        );
+    }
+}
+
 #[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")];
+    let all_buckets = vec![bucket_info_with_created_time("stable"), bucket_info_with_created_time("new")];
 
     assert!(
         prepare_scoped_set_scan(
@@ -1067,7 +1111,7 @@ fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
 #[test]
 fn scoped_set_scan_requires_an_exact_complete_baseline() {
     let baseline_digest = DataUsageScanPlanDigest([5; 32]);
-    let all_buckets = vec![bucket_info("dirty")];
+    let all_buckets = vec![bucket_info_with_created_time("dirty")];
     let scope = ScannerBucketScanScope {
         selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
         baseline_scan_plan_digest: Some(baseline_digest),
@@ -1088,6 +1132,10 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() {
     not_durable.info.last_update = None;
     assert!(prepare_scoped_set_scan(&not_durable, &all_buckets, &all_buckets, &scope, generation).is_none());
 
+    let mut unscoped_usage = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    unscoped_usage.cache.get_mut(DATA_USAGE_ROOT).expect("set root").objects = 1;
+    assert!(prepare_scoped_set_scan(&unscoped_usage, &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());
@@ -1098,6 +1146,13 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() {
     };
     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());
+    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &scope, generation).is_some());
+
+    let unidentified_buckets = vec![bucket_info("dirty")];
+    assert!(
+        prepare_scoped_set_scan(&complete, &unidentified_buckets, &unidentified_buckets, &scope, generation).is_some(),
+        "fully selected buckets are rebuilt without reusing an unproven incarnation"
+    );
 
     let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
     future_cache.info.next_cycle = generation.want_cycle.saturating_add(1);