From 2d159635eddb9f17ad504410be91e20557eee4fe Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Sat, 5 Sep 2026 16:32:02 +0800 Subject: [PATCH] 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 Co-authored-by: cxymds Co-authored-by: Zhengchao An --- crates/scanner/src/scanner.rs | 23 ++- crates/scanner/src/scanner/activity.rs | 41 ++++ crates/scanner/src/scanner/tests.rs | 38 ++++ crates/scanner/src/scanner_io.rs | 119 ++++++++++- crates/scanner/src/scanner_io/cache.rs | 18 ++ crates/scanner/src/scanner_io/io_cycle.rs | 95 ++++++++- .../src/scanner_io/publish_gate_tests.rs | 22 ++ crates/scanner/src/scanner_io/tests.rs | 190 ++++++++++++++++++ crates/scanner/src/storage_api.rs | 4 +- 9 files changed, 539 insertions(+), 11 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index fcf67eed0..0f97924a4 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1703,14 +1703,18 @@ where let (sender, receiver) = mpsc::channel::(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::{ diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index 2f8aed1b5..ffcbc5313 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/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(); diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index efe29f612..244e9088f 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/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([ diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 4807e11b8..c8353bae9 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/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, } +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, 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, + 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, + peer_snapshots: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>, +) -> Option> { + 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 { + let data = proof.data?; + let baseline = serde_json::from_slice::(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, + 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::>(); + let selected_buckets = dirty_buckets + .into_iter() + .filter(|bucket| current_buckets.contains(bucket.as_str())) + .collect::>(); + 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; diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 9522ffed2..7ff684c99 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/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::>>()?; + 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)) diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 3b14e9ff2..2947d638d 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/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, +} + +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( + 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::>(); + 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(store: &S, request: ScannerCycleRequest) -> Result @@ -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::>() == *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; diff --git a/crates/scanner/src/scanner_io/publish_gate_tests.rs b/crates/scanner/src/scanner_io/publish_gate_tests.rs index d740cb2eb..f131cf437 100644 --- a/crates/scanner/src/scanner_io/publish_gate_tests.rs +++ b/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![ + (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 diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 6fddafdfe..ec2c1ad65 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/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::(&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::(&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]); diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index a0aa874b0..f81065d59 100644 --- a/crates/scanner/src/storage_api.rs +++ b/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::{