feat(scanner): prepare scoped bucket cache scans (#7136)

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

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;