feat(scanner): plan dirty bucket cache refreshes (#7146)

* feat(scanner): plan dirty bucket cache refreshes

* fix(scanner): route peer snapshot through storage boundary

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: cxymds <cxymds@gmail.com>
Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

crates/scanner/src/scanner.rs

@@ -1703,14 +1703,18 @@ where
     let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
 
     let done_cycle = Metrics::time(Metric::ScanCycle);
-    let scan_result = crate::scanner_io::nsscanner_with_storage_status(
+    let scan_result = crate::scanner_io::nsscanner_with_storage_status_scoped(
         storeapi.as_ref(),
-        cycle_budget.token(),
-        cycle_budget.clone(),
-        sender,
-        cycle_info.current,
-        leader_epoch,
-        scan_mode,
+        crate::scanner_io::ScannerCycleRequest {
+            ctx: cycle_budget.token(),
+            budget: cycle_budget.clone(),
+            updates: sender,
+            want_cycle: cycle_info.current,
+            leader_epoch,
+            scan_mode,
+            scan_scope: crate::scanner_io::ScannerBucketScanScope::default(),
+            persisted_usage_baseline: usage_persist_baseline.data.clone(),
+        },
     )
     .await;
     let publication_defer_reason = match &scan_result {
@@ -3424,10 +3428,13 @@ use cycle_state::*;
 use leadership::*;
 use usage_store::*;
 
+#[cfg(test)]
+pub(crate) use activity::scanner_activity_snapshot_digest;
 pub use activity::scanner_topology_digest;
 pub(crate) use activity::{
     ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, probe_scanner_activity, scanner_activity_allows_usage_publication,
-    scanner_activity_publication_lease_targets, scanner_activity_snapshot_digest, scanner_dirty_usage_acknowledgements,
+    scanner_activity_dirty_usage_state_for_host, scanner_activity_publication_lease_targets, scanner_activity_structural_digest,
+    scanner_dirty_usage_acknowledgements,
 };
 pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
 pub use backlog::{

crates/scanner/src/scanner/activity.rs

@@ -902,6 +902,7 @@ where
     observation
 }
 
+#[cfg(test)]
 pub(crate) fn scanner_activity_snapshot_digest(snapshot: &ScannerActivitySnapshot) -> [u8; 32] {
     let mut hasher = Sha256::new();
     hasher.update(u64::try_from(snapshot.len()).unwrap_or(u64::MAX).to_be_bytes());
@@ -925,6 +926,30 @@ pub(crate) fn scanner_activity_snapshot_digest(snapshot: &ScannerActivitySnapsho
     hasher.finalize().into()
 }
 
+/// Hash the activity inputs that make an existing scanner cache unsafe to
+/// reuse. Regular namespace writes and dirty-usage generations are omitted:
+/// their affected buckets are tracked separately and may be refreshed from a
+/// complete authoritative cache baseline.
+pub(crate) fn scanner_activity_structural_digest(snapshot: &ScannerActivitySnapshot) -> [u8; 32] {
+    let mut hasher = Sha256::new();
+    hasher.update(u64::try_from(snapshot.len()).unwrap_or(u64::MAX).to_be_bytes());
+    for (host, activity) in snapshot {
+        let host = host.as_bytes();
+        let instance_id = activity.instance_id.as_bytes();
+        hasher.update(u64::try_from(host.len()).unwrap_or(u64::MAX).to_be_bytes());
+        hasher.update(host);
+        hasher.update(u64::try_from(instance_id.len()).unwrap_or(u64::MAX).to_be_bytes());
+        hasher.update(instance_id);
+        hasher.update(activity.maintenance_generation.to_be_bytes());
+        hasher.update(activity.protocol_version.to_be_bytes());
+        hasher.update(activity.topology_digest);
+        hasher.update([u8::from(activity.data_movement_active)]);
+        hasher.update(activity.movement_generation.to_be_bytes());
+        hasher.update([u8::from(activity.publication_blocked)]);
+    }
+    hasher.finalize().into()
+}
+
 pub(crate) fn scanner_activity_allows_usage_publication(snapshot: &ScannerActivitySnapshot) -> bool {
     !snapshot.is_empty()
         && snapshot.values().all(|activity| {
@@ -955,6 +980,22 @@ pub(crate) fn scanner_dirty_usage_acknowledgements(snapshot: &ScannerActivitySna
         .collect()
 }
 
+pub(crate) fn scanner_activity_dirty_usage_state_for_host<'a>(
+    snapshot: &'a ScannerActivitySnapshot,
+    host: &str,
+) -> Option<(&'a str, u64, bool)> {
+    snapshot
+        .get(host)
+        .filter(|_| host != LOCAL_SCANNER_ACTIVITY_NODE)
+        .map(|activity| {
+            (
+                activity.instance_id.as_str(),
+                activity.dirty_usage_generation,
+                activity.dirty_usage_pending,
+            )
+        })
+}
+
 pub fn scanner_topology_digest(storeapi: &ECStore) -> [u8; 32] {
     let endpoint_pools = storeapi.endpoints();
     let mut hasher = Sha256::new();

crates/scanner/src/scanner/tests.rs

@@ -8169,6 +8169,44 @@ fn scanner_activity_snapshot_digest_fences_dirty_usage_state() {
     assert_ne!(scanner_activity_snapshot_digest(&clean), scanner_activity_snapshot_digest(&pending));
 }
 
+#[test]
+fn scanner_activity_structural_digest_ignores_regular_bucket_writes() {
+    let baseline = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
+    let mut written = baseline.clone();
+    let activity = written.get_mut("node-2").expect("node should exist");
+    activity.namespace_generation = 8;
+    activity.dirty_usage_generation = 6;
+    activity.dirty_usage_pending = true;
+
+    assert_ne!(scanner_activity_snapshot_digest(&baseline), scanner_activity_snapshot_digest(&written));
+    assert_eq!(
+        scanner_activity_structural_digest(&baseline),
+        scanner_activity_structural_digest(&written),
+        "bucket writes are refreshed through the dirty-bucket scope rather than invalidating every cache"
+    );
+}
+
+#[test]
+fn scanner_activity_structural_digest_fences_restart_and_maintenance() {
+    let baseline = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
+    let mut restarted = baseline.clone();
+    restarted.get_mut("node-2").expect("node should exist").instance_id = "epoch-b".to_string();
+    let mut maintained = baseline.clone();
+    maintained
+        .get_mut("node-2")
+        .expect("node should exist")
+        .maintenance_generation = 4;
+
+    assert_ne!(
+        scanner_activity_structural_digest(&baseline),
+        scanner_activity_structural_digest(&restarted)
+    );
+    assert_ne!(
+        scanner_activity_structural_digest(&baseline),
+        scanner_activity_structural_digest(&maintained)
+    );
+}
+
 #[test]
 fn scanner_dirty_usage_acknowledgements_exclude_local_and_clean_nodes() {
     let snapshot = BTreeMap::from([

crates/scanner/src/scanner_io.rs

@@ -21,6 +21,7 @@ use crate::{
     DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo, DataUsageScanPlanDigest, DataUsageSnapshotSetState,
     ScannerError, SizeSummary, TierStats,
 };
+use bytes::Bytes;
 use futures::future::join_all;
 use metrics::counter;
 use rand::seq::SliceRandom as _;
@@ -54,6 +55,7 @@ use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, warn};
 
 use crate::ScannerObjectInfo as ObjectInfo;
+use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
 use crate::storage_api::ScannerStorage;
 use crate::storage_api::scan::NamespaceLocking as _;
 use crate::storage_api::scanner_io::{BucketInfo, BucketOptions};
@@ -111,6 +113,121 @@ pub(crate) struct ScannerBucketScanScope {
     baseline_scan_plan_digest: Option<DataUsageScanPlanDigest>,
 }
 
+impl ScannerBucketScanScope {
+    fn is_default(&self) -> bool {
+        self.selected_buckets.is_none() && self.baseline_scan_plan_digest.is_none()
+    }
+
+    fn from_dirty_buckets(selected_buckets: HashSet<String>, baseline_scan_plan_digest: DataUsageScanPlanDigest) -> Self {
+        Self {
+            selected_buckets: Some(Arc::new(selected_buckets)),
+            baseline_scan_plan_digest: Some(baseline_scan_plan_digest),
+        }
+    }
+}
+
+#[derive(Clone, Copy)]
+pub(super) struct ScannerCacheBaselineProof<'a> {
+    pub(super) data: Option<&'a Bytes>,
+    pub(super) expected_sources: &'a HashSet<DataUsageCacheSource>,
+    pub(super) leader_epoch: u64,
+    pub(super) want_cycle: u64,
+    pub(super) scan_plan_digest: DataUsageScanPlanDigest,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct ScannerPeerDirtyUsageExpectation {
+    instance_id: String,
+    generation: u64,
+    pending: bool,
+}
+
+fn verified_remote_dirty_usage_buckets(
+    expected_peers: &HashMap<String, ScannerPeerDirtyUsageExpectation>,
+    peer_snapshots: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>,
+) -> Option<HashSet<String>> {
+    if expected_peers.is_empty() || peer_snapshots.len() != expected_peers.len() {
+        return None;
+    }
+
+    let mut received_peers = HashSet::with_capacity(peer_snapshots.len());
+    let mut dirty_buckets = HashSet::new();
+    for (host, snapshot) in peer_snapshots {
+        let expected = expected_peers.get(&host)?;
+        if !received_peers.insert(host)
+            || snapshot.instance_id != expected.instance_id
+            || snapshot.generation != expected.generation
+            || snapshot.generation == u64::MAX
+            || snapshot.protocol_version != crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION
+            || !snapshot.complete
+            || snapshot.pending_bucket_count != u64::try_from(snapshot.buckets.len()).unwrap_or(u64::MAX)
+            || (expected.pending && snapshot.pending_bucket_count == 0)
+        {
+            return None;
+        }
+        dirty_buckets.extend(snapshot.buckets.into_keys());
+    }
+
+    (received_peers.len() == expected_peers.len()).then_some(dirty_buckets)
+}
+
+fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
+    let data = proof.data?;
+    let baseline = serde_json::from_slice::<DataUsageInfo>(data).ok()?;
+    if !baseline.is_complete_bucket_usage_snapshot()
+        || baseline.usage_snapshot_partial
+        || baseline.usage_snapshot_converged != Some(true)
+        || baseline.scanner_epoch != Some(proof.leader_epoch)
+        || baseline.usage_snapshot_set_states.len() != proof.expected_sources.len()
+    {
+        return None;
+    }
+
+    let mut states = HashSet::with_capacity(baseline.usage_snapshot_set_states.len());
+    for state in &baseline.usage_snapshot_set_states {
+        let source = DataUsageCacheSource::new(usize::try_from(state.pool_index).ok()?, usize::try_from(state.set_index).ok()?);
+        if !proof.expected_sources.contains(&source)
+            || !states.insert(source)
+            || !state.complete
+            || state.tombstone
+            || state.scanner_epoch != Some(proof.leader_epoch)
+            || state.scanner_cycle.is_none_or(|cycle| cycle > proof.want_cycle)
+            || state.scan_plan_digest != Some(proof.scan_plan_digest.0)
+        {
+            return None;
+        }
+    }
+
+    (states == *proof.expected_sources).then_some(proof.scan_plan_digest)
+}
+
+fn scoped_scan_scope_from_dirty_buckets(
+    requested_scope: ScannerBucketScanScope,
+    dirty_buckets: HashSet<String>,
+    dirty_snapshot_complete: bool,
+    all_buckets: &[BucketInfo],
+    baseline_proof: ScannerCacheBaselineProof<'_>,
+) -> ScannerBucketScanScope {
+    if !requested_scope.is_default() || !dirty_snapshot_complete {
+        return requested_scope;
+    }
+
+    let current_buckets = all_buckets.iter().map(|bucket| bucket.name.as_str()).collect::<HashSet<_>>();
+    let selected_buckets = dirty_buckets
+        .into_iter()
+        .filter(|bucket| current_buckets.contains(bucket.as_str()))
+        .collect::<HashSet<_>>();
+    if selected_buckets.is_empty() {
+        return requested_scope;
+    }
+
+    let Some(baseline_scan_plan_digest) = complete_scanner_cache_baseline_plan_digest(baseline_proof) else {
+        return requested_scope;
+    };
+
+    ScannerBucketScanScope::from_dirty_buckets(selected_buckets, baseline_scan_plan_digest)
+}
+
 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))
 }
@@ -749,7 +866,7 @@ 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;
+pub(crate) use io_cycle::{ScannerCycleRequest, nsscanner_with_storage_status_scoped};
 mod io_disk;
 #[cfg(test)]
 mod publish_gate_tests;

crates/scanner/src/scanner_io/cache.rs

@@ -282,9 +282,26 @@ pub(super) fn completed_data_usage_info(
         .iter()
         .map(|(bucket, usage)| (bucket.clone(), usage.size))
         .collect();
+    let mut usage_snapshot_set_states = results
+        .iter()
+        .map(|result| {
+            let source = result.info.source?;
+            Some(DataUsageSnapshotSetState {
+                pool_index: u64::try_from(source.pool_index).ok()?,
+                set_index: u64::try_from(source.set_index).ok()?,
+                scanner_cycle: Some(result.info.next_cycle),
+                scanner_epoch: Some(result.info.leader_epoch),
+                scan_plan_digest: Some(result.info.scan_plan_digest?.0),
+                complete: true,
+                tombstone: false,
+            })
+        })
+        .collect::<Option<Vec<_>>>()?;
+    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),
         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()?,
@@ -295,6 +312,7 @@ pub(super) fn completed_data_usage_info(
         bucket_sizes,
         buckets_usage,
         usage_snapshot_complete: true,
+        usage_snapshot_set_states,
         ..Default::default()
     };
     Some((data_usage_info, merged_last_update))

crates/scanner/src/scanner_io/io_cycle.rs

@@ -71,6 +71,7 @@ where
         leader_epoch,
         scan_mode,
         scan_scope: ScannerBucketScanScope::default(),
+        persisted_usage_baseline: None,
     };
     nsscanner_with_storage_status_scoped(store, request).await
 }
@@ -83,6 +84,79 @@ pub(crate) struct ScannerCycleRequest {
     pub(crate) leader_epoch: u64,
     pub(crate) scan_mode: HealScanMode,
     pub(crate) scan_scope: ScannerBucketScanScope,
+    pub(crate) persisted_usage_baseline: Option<Bytes>,
+}
+
+struct ScannerBucketScopeResolution<'a> {
+    requested_scope: ScannerBucketScanScope,
+    baseline_proof: ScannerCacheBaselineProof<'a>,
+    activity_before: &'a crate::scanner::ScannerActivitySnapshot,
+    dirty_usage_snapshot: &'a DirtyUsageSnapshot,
+    all_buckets: &'a [BucketInfo],
+}
+
+async fn resolve_scanner_bucket_scan_scope<S>(
+    store: &S,
+    distributed: bool,
+    resolution: ScannerBucketScopeResolution<'_>,
+) -> ScannerBucketScanScope
+where
+    S: ScannerStorage,
+{
+    if !resolution.requested_scope.is_default()
+        || !resolution.dirty_usage_snapshot.covers_all_pending
+        || resolution.dirty_usage_snapshot.generation == u64::MAX
+        || resolution.dirty_usage_snapshot.buckets.len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES
+    {
+        return resolution.requested_scope;
+    }
+
+    let mut dirty_buckets = resolution
+        .dirty_usage_snapshot
+        .buckets
+        .keys()
+        .cloned()
+        .collect::<HashSet<_>>();
+    if distributed {
+        let Some(notification_system) = store.scanner_notification_system() else {
+            return resolution.requested_scope;
+        };
+        let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else {
+            return resolution.requested_scope;
+        };
+        let mut expected_peers = HashMap::new();
+        for (host, lease_instance_id, _) in crate::scanner::scanner_activity_publication_lease_targets(resolution.activity_before)
+        {
+            let Some((activity_instance_id, generation, pending)) =
+                crate::scanner::scanner_activity_dirty_usage_state_for_host(resolution.activity_before, &host)
+            else {
+                return resolution.requested_scope;
+            };
+            if activity_instance_id != lease_instance_id || expected_peers.contains_key(&host) {
+                return resolution.requested_scope;
+            }
+            expected_peers.insert(
+                host,
+                ScannerPeerDirtyUsageExpectation {
+                    instance_id: activity_instance_id.to_string(),
+                    generation,
+                    pending,
+                },
+            );
+        }
+        let Some(remote_dirty_buckets) = verified_remote_dirty_usage_buckets(&expected_peers, peer_snapshots) else {
+            return resolution.requested_scope;
+        };
+        dirty_buckets.extend(remote_dirty_buckets);
+    }
+
+    scoped_scan_scope_from_dirty_buckets(
+        resolution.requested_scope,
+        dirty_buckets,
+        true,
+        resolution.all_buckets,
+        resolution.baseline_proof,
+    )
 }
 
 pub(crate) async fn nsscanner_with_storage_status_scoped<S>(store: &S, request: ScannerCycleRequest) -> Result<ScannerCycleResult>
@@ -97,6 +171,7 @@ where
         leader_epoch,
         scan_mode,
         scan_scope,
+        persisted_usage_baseline,
     } = request;
     let child_token = ctx.child_token();
     let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
@@ -186,8 +261,26 @@ where
     }
     bucket_plan_complete &= buckets_by_source.keys().copied().collect::<HashSet<_>>() == *expected_sources;
     let scan_plan_digest =
-        scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_snapshot_digest(&activity_before));
+        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));
+    let scan_scope = resolve_scanner_bucket_scan_scope(
+        store,
+        distributed,
+        ScannerBucketScopeResolution {
+            requested_scope: scan_scope,
+            baseline_proof: ScannerCacheBaselineProof {
+                data: persisted_usage_baseline.as_ref(),
+                expected_sources: &expected_sources,
+                leader_epoch,
+                want_cycle,
+                scan_plan_digest,
+            },
+            activity_before: &activity_before,
+            dirty_usage_snapshot: &dirty_usage_snapshot,
+            all_buckets: &all_buckets,
+        },
+    )
+    .await;
     let cache_cycle_floor = Arc::new(AtomicU64::new(want_cycle));
     let tier_registry = runtime_tier_registry_for_cycle(want_cycle, leader_epoch).await;
     let tier_registry_generation = tier_registry.generation;

crates/scanner/src/scanner_io/publish_gate_tests.rs

@@ -655,9 +655,31 @@ fn completed_data_usage_info_requires_every_set_before_publish() {
         .expect("all completed sets should produce a publishable data usage snapshot");
     assert_eq!(last_update, SystemTime::UNIX_EPOCH + Duration::from_secs(20));
     assert_eq!(data_usage_info.scanner_cycle, Some(0));
+    assert_eq!(data_usage_info.scanner_epoch, Some(0));
     assert_eq!(data_usage_info.objects_total_count, 3);
     assert_eq!(data_usage_info.buckets_usage.len(), 3);
     assert!(data_usage_info.usage_snapshot_complete);
+    assert_eq!(
+        data_usage_info
+            .usage_snapshot_set_states
+            .iter()
+            .map(|state| {
+                (
+                    state.pool_index,
+                    state.set_index,
+                    state.scanner_cycle,
+                    state.scanner_epoch,
+                    state.scan_plan_digest,
+                    state.complete,
+                    state.tombstone,
+                )
+            })
+            .collect::<Vec<_>>(),
+        vec![
+            (0, 0, Some(0), Some(0), Some(TEST_PLAN_DIGEST.0), true, false),
+            (1, 0, Some(0), Some(0), Some(TEST_PLAN_DIGEST.0), true, false),
+        ]
+    );
     assert_eq!(
         data_usage_info
             .buckets_usage

crates/scanner/src/scanner_io/tests.rs

@@ -17,6 +17,7 @@ use super::io_disk::tier_stats_template;
 use super::*;
 use crate::scanner_budget::ScannerCycleBudgetConfig;
 use crate::scanner_folder::ScannerItem;
+use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
 use crate::storage_api::owner::{
     EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats,
 };
@@ -796,6 +797,195 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa
     cache
 }
 
+fn complete_usage_baseline(
+    source: DataUsageCacheSource,
+    scan_plan_digest: DataUsageScanPlanDigest,
+    scanner_cycle: u64,
+    scanner_epoch: u64,
+) -> bytes::Bytes {
+    let baseline = DataUsageInfo {
+        last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10)),
+        scanner_cycle: Some(scanner_cycle),
+        scanner_epoch: Some(scanner_epoch),
+        buckets_count: 1,
+        buckets_usage: HashMap::from([("photos".to_string(), Default::default())]),
+        usage_snapshot_complete: true,
+        usage_snapshot_converged: Some(true),
+        usage_snapshot_set_states: vec![DataUsageSnapshotSetState {
+            pool_index: u64::try_from(source.pool_index).expect("test pool index should fit"),
+            set_index: u64::try_from(source.set_index).expect("test set index should fit"),
+            scanner_cycle: Some(scanner_cycle),
+            scanner_epoch: Some(scanner_epoch),
+            scan_plan_digest: Some(scan_plan_digest.0),
+            complete: true,
+            tombstone: false,
+        }],
+        ..Default::default()
+    };
+    bytes::Bytes::from(serde_json::to_vec(&baseline).expect("test baseline should encode"))
+}
+
+#[test]
+fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance() {
+    let source = DataUsageCacheSource::new(1, 2);
+    let expected_sources = HashSet::from([source]);
+    let scan_plan_digest = DataUsageScanPlanDigest([9; 32]);
+    let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11);
+
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
+            data: Some(&baseline),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        }),
+        Some(scan_plan_digest)
+    );
+
+    let mut incomplete = serde_json::from_slice::<DataUsageInfo>(&baseline).expect("test baseline should decode");
+    incomplete.usage_snapshot_converged = Some(false);
+    let incomplete = bytes::Bytes::from(serde_json::to_vec(&incomplete).expect("test baseline should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
+            data: Some(&incomplete),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        }),
+        None
+    );
+
+    let mut wrong_provenance = serde_json::from_slice::<DataUsageInfo>(&baseline).expect("test baseline should decode");
+    wrong_provenance.usage_snapshot_set_states[0].scan_plan_digest = Some([8; 32]);
+    let wrong_provenance = bytes::Bytes::from(serde_json::to_vec(&wrong_provenance).expect("test baseline should encode"));
+    assert_eq!(
+        complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
+            data: Some(&wrong_provenance),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest,
+        }),
+        None
+    );
+}
+
+#[test]
+fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() {
+    let source = DataUsageCacheSource::new(1, 2);
+    let expected_sources = HashSet::from([source]);
+    let baseline_scan_plan_digest = DataUsageScanPlanDigest([4; 32]);
+    let current_scan_plan_digest = DataUsageScanPlanDigest([5; 32]);
+    let baseline = complete_usage_baseline(source, current_scan_plan_digest, 7, 11);
+    let scope = scoped_scan_scope_from_dirty_buckets(
+        ScannerBucketScanScope::default(),
+        HashSet::from(["photos".to_string(), "deleted".to_string()]),
+        true,
+        &[bucket_info("photos")],
+        ScannerCacheBaselineProof {
+            data: Some(&baseline),
+            expected_sources: &expected_sources,
+            leader_epoch: 11,
+            want_cycle: 8,
+            scan_plan_digest: current_scan_plan_digest,
+        },
+    );
+
+    assert_eq!(scope.baseline_scan_plan_digest, Some(current_scan_plan_digest));
+    assert_eq!(
+        scope
+            .selected_buckets
+            .as_deref()
+            .expect("validated scope should select a bucket"),
+        &HashSet::from(["photos".to_string()])
+    );
+    assert_ne!(scope.baseline_scan_plan_digest, Some(baseline_scan_plan_digest));
+}
+
+fn peer_dirty_usage_snapshot(
+    instance_id: &str,
+    generation: u64,
+    complete: bool,
+    buckets: &[(&str, u64)],
+) -> EcstoreScannerPeerDirtyUsageSnapshot {
+    EcstoreScannerPeerDirtyUsageSnapshot {
+        instance_id: instance_id.to_string(),
+        generation,
+        pending_bucket_count: u64::try_from(buckets.len()).expect("test bucket count should fit"),
+        protocol_version: crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION,
+        complete,
+        buckets: buckets
+            .iter()
+            .map(|(bucket, generation)| ((*bucket).to_string(), *generation))
+            .collect(),
+    }
+}
+
+#[test]
+fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots() {
+    let expected_peers = HashMap::from([
+        (
+            "node-a:9000".to_string(),
+            ScannerPeerDirtyUsageExpectation {
+                instance_id: "instance-a".to_string(),
+                generation: 7,
+                pending: true,
+            },
+        ),
+        (
+            "node-b:9000".to_string(),
+            ScannerPeerDirtyUsageExpectation {
+                instance_id: "instance-b".to_string(),
+                generation: 3,
+                pending: false,
+            },
+        ),
+    ]);
+
+    assert_eq!(
+        verified_remote_dirty_usage_buckets(
+            &expected_peers,
+            vec![
+                (
+                    "node-a:9000".to_string(),
+                    peer_dirty_usage_snapshot("instance-a", 7, true, &[("photos", 7)]),
+                ),
+                (
+                    "node-b:9000".to_string(),
+                    peer_dirty_usage_snapshot("instance-b", 3, true, &[("archive", 3)]),
+                ),
+            ],
+        ),
+        Some(HashSet::from(["photos".to_string(), "archive".to_string()]))
+    );
+}
+
+#[test]
+fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state() {
+    let expected_peers = HashMap::from([(
+        "node-a:9000".to_string(),
+        ScannerPeerDirtyUsageExpectation {
+            instance_id: "instance-a".to_string(),
+            generation: 7,
+            pending: true,
+        },
+    )]);
+
+    for snapshot in [
+        peer_dirty_usage_snapshot("instance-a", 7, false, &[("photos", 7)]),
+        peer_dirty_usage_snapshot("instance-a", 6, true, &[("photos", 6)]),
+        peer_dirty_usage_snapshot("instance-b", 7, true, &[("photos", 7)]),
+        peer_dirty_usage_snapshot("instance-a", 7, true, &[]),
+    ] {
+        assert!(
+            verified_remote_dirty_usage_buckets(&expected_peers, vec![("node-a:9000".to_string(), snapshot)]).is_none(),
+            "incomplete, stale, mismatched, or empty pending peer state must fall back to a full scan"
+        );
+    }
+}
+
 #[test]
 fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
     let baseline_digest = DataUsageScanPlanDigest([1; 32]);

crates/scanner/src/storage_api.rs

@@ -103,7 +103,9 @@ pub(crate) use rustfs_ecstore::api::rebalance::{
     RebalStatus as EcstoreRebalStatus, RebalanceInfo as EcstoreRebalanceInfo, RebalanceMeta as EcstoreRebalanceMeta,
     RebalanceStats as EcstoreRebalanceStats,
 };
-pub(crate) use rustfs_ecstore::api::rpc::ScannerBucketListing as EcstoreScannerBucketListing;
+pub(crate) use rustfs_ecstore::api::rpc::{
+    ScannerBucketListing as EcstoreScannerBucketListing, ScannerPeerDirtyUsageSnapshot as EcstoreScannerPeerDirtyUsageSnapshot,
+};
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::runtime::InstanceContext as EcstoreInstanceContext;
 pub(crate) use rustfs_ecstore::api::runtime::{