diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 8cc4c481e..03f87a61f 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -3573,6 +3573,8 @@ use leadership::*; pub(crate) use usage_store::RootPublicationProof; use usage_store::*; +#[cfg(test)] +pub(crate) use activity::scanner_node_activity_for_tests; pub use activity::scanner_topology_digest; pub(crate) use activity::{ ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, ScannerDirtyUsageAcknowledgementKind, probe_scanner_activity, diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index 1b2761bb1..582649147 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -441,6 +441,27 @@ pub(crate) struct ScannerNodeActivity { pub(crate) type ScannerActivitySnapshot = BTreeMap; +#[cfg(test)] +pub(crate) fn scanner_node_activity_for_tests( + instance_id: &str, + namespace_generation: u64, + dirty_usage_generation: u64, + dirty_usage_pending: bool, +) -> ScannerNodeActivity { + ScannerNodeActivity { + instance_id: instance_id.to_string(), + namespace_generation, + maintenance_generation: 0, + protocol_version: SCANNER_ACTIVITY_PROTOCOL_VERSION, + topology_digest: [0; 32], + data_movement_active: false, + dirty_usage_generation, + dirty_usage_pending, + movement_generation: 0, + publication_blocked: false, + } +} + #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct ScannerDirtyUsageAcknowledgement { pub(crate) host: String, diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 1f7cd37e6..af6b4ac58 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -165,6 +165,11 @@ struct VerifiedRemoteDirtyUsage { acknowledgements: Vec, } +struct ScannerBucketScopeResolutionResult { + scope: ScannerBucketScanScope, + remote_dirty_usage_acknowledgements: Vec, +} + fn verified_remote_dirty_usage( expected_peers: &HashMap, peer_snapshots: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>, @@ -229,6 +234,61 @@ fn scanner_scoped_dirty_usage_ack_exceeds_cost_threshold( }) } +fn resolve_remote_dirty_usage_scope( + requested_scope: ScannerBucketScanScope, + mut dirty_buckets: HashSet, + remote_dirty_usage: VerifiedRemoteDirtyUsage, + all_buckets: &[BucketInfo], + baseline_proof: ScannerCacheBaselineProof<'_>, +) -> ScannerBucketScopeResolutionResult { + let default_result = |scope: ScannerBucketScanScope| ScannerBucketScopeResolutionResult { + scope, + remote_dirty_usage_acknowledgements: Vec::new(), + }; + + dirty_buckets.extend(remote_dirty_usage.dirty_buckets); + // Peer snapshots contribute bucket names only; the local prefix scopes + // would narrow a bucket a peer dirtied elsewhere, so the merged scope + // stays at bucket granularity (same rule as the local fallthrough). + let scope = scoped_scan_scope_from_dirty_buckets(requested_scope, dirty_buckets, None, true, all_buckets, baseline_proof); + if scope.is_default() { + return default_result(scope); + } + let Some(selected_buckets) = scope.selected_buckets.as_ref() else { + return default_result(scope); + }; + let mut scoped_acknowledgements = Vec::with_capacity(remote_dirty_usage.acknowledgements.len()); + for acknowledgement in remote_dirty_usage.acknowledgements { + let crate::scanner::ScannerDirtyUsageAcknowledgement { + host, + instance_id, + kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries }, + } = acknowledgement + else { + return default_result(scope); + }; + let entries = entries + .into_iter() + .filter(|entry| selected_buckets.contains(&entry.bucket)) + .collect::>(); + if !entries.is_empty() { + scoped_acknowledgements.push(crate::scanner::ScannerDirtyUsageAcknowledgement { + host, + instance_id, + kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries }, + }); + } + } + if scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&scoped_acknowledgements) { + return default_result(ScannerBucketScanScope::default()); + } + + ScannerBucketScopeResolutionResult { + scope, + remote_dirty_usage_acknowledgements: scoped_acknowledgements, + } +} + fn complete_scanner_cache_snapshot_plan_digest( snapshot: &DataUsageInfo, proof: ScannerCacheBaselineProof<'_>, diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 1cc857f46..a396248d4 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -98,18 +98,17 @@ pub(crate) struct ScannerCycleRequest { pub(crate) resolved_scope_observer: 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], - requires_full_scan: bool, -} - -struct ScannerBucketScopeResolutionResult { - scope: ScannerBucketScanScope, - remote_dirty_usage_acknowledgements: Vec, +pub(super) struct ScannerBucketScopeResolution<'a> { + pub(super) requested_scope: ScannerBucketScanScope, + pub(super) baseline_proof: ScannerCacheBaselineProof<'a>, + pub(super) activity_before: &'a crate::scanner::ScannerActivitySnapshot, + pub(super) dirty_usage_snapshot: &'a DirtyUsageSnapshot, + pub(super) all_buckets: &'a [BucketInfo], + pub(super) requires_full_scan: bool, + #[cfg(test)] + pub(super) test_peer_snapshots: Option>, + #[cfg(test)] + pub(super) test_scoped_dirty_usage_capability: Option, } async fn resolve_scanner_bucket_scan_scope( @@ -135,18 +134,35 @@ where return default_result(resolution.requested_scope); } - let mut dirty_buckets = resolution + let dirty_buckets = resolution .dirty_usage_snapshot .buckets .keys() .cloned() .collect::>(); if distributed { - let Some(notification_system) = store.scanner_notification_system() else { - return default_result(resolution.requested_scope); + let notification_system = store.scanner_notification_system(); + #[cfg(test)] + let peer_snapshots = if let Some(peer_snapshots) = resolution.test_peer_snapshots.clone() { + peer_snapshots + } else { + let Some(notification_system) = notification_system.as_ref() else { + return default_result(resolution.requested_scope); + }; + let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else { + return default_result(resolution.requested_scope); + }; + peer_snapshots }; - let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else { - return default_result(resolution.requested_scope); + #[cfg(not(test))] + let peer_snapshots = { + let Some(notification_system) = notification_system.as_ref() else { + return default_result(resolution.requested_scope); + }; + let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else { + return default_result(resolution.requested_scope); + }; + peer_snapshots }; let mut expected_peers = HashMap::new(); for (host, lease_instance_id, _) in crate::scanner::scanner_activity_publication_lease_targets(resolution.activity_before) @@ -171,68 +187,57 @@ where let Some(remote_dirty_usage) = verified_remote_dirty_usage(&expected_peers, peer_snapshots) else { return default_result(resolution.requested_scope); }; - dirty_buckets.extend(remote_dirty_usage.dirty_buckets); - // Peer snapshots contribute bucket names only; the local prefix scopes - // would narrow a bucket a peer dirtied elsewhere, so the merged scope - // stays at bucket granularity (same rule as the local fallthrough). - let scope = scoped_scan_scope_from_dirty_buckets( + let remote_resolution = resolve_remote_dirty_usage_scope( resolution.requested_scope, dirty_buckets, - None, - true, + remote_dirty_usage, resolution.all_buckets, resolution.baseline_proof, ); - if scope.is_default() { - return default_result(scope); - } - let Some(selected_buckets) = scope.selected_buckets.as_ref() else { - return default_result(scope); - }; - let mut scoped_acknowledgements = Vec::with_capacity(remote_dirty_usage.acknowledgements.len()); - for acknowledgement in remote_dirty_usage.acknowledgements { - let crate::scanner::ScannerDirtyUsageAcknowledgement { - host, - instance_id, - kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries }, - } = acknowledgement - else { - return default_result(scope); + if !remote_resolution.remote_dirty_usage_acknowledgements.is_empty() { + #[cfg(test)] + let capability_supported = if let Some(capability_supported) = resolution.test_scoped_dirty_usage_capability { + capability_supported + } else { + let Some(notification_system) = notification_system.as_ref() else { + return default_result(ScannerBucketScanScope::default()); + }; + let capability_acknowledgements = remote_resolution + .remote_dirty_usage_acknowledgements + .clone() + .into_iter() + .map(Into::into) + .collect::>(); + matches!( + notification_system + .scanner_scoped_dirty_usage_capabilities(capability_acknowledgements) + .await, + Ok(true) + ) }; - let entries = entries - .into_iter() - .filter(|entry| selected_buckets.contains(&entry.bucket)) - .collect::>(); - if !entries.is_empty() { - scoped_acknowledgements.push(crate::scanner::ScannerDirtyUsageAcknowledgement { - host, - instance_id, - kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries }, - }); - } - } - if super::scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&scoped_acknowledgements) { - return default_result(ScannerBucketScanScope::default()); - } - if !scoped_acknowledgements.is_empty() { - let capability_acknowledgements = scoped_acknowledgements - .clone() - .into_iter() - .map(Into::into) - .collect::>(); - if !matches!( - notification_system - .scanner_scoped_dirty_usage_capabilities(capability_acknowledgements) - .await, - Ok(true) - ) { + #[cfg(not(test))] + let capability_supported = { + let Some(notification_system) = notification_system.as_ref() else { + return default_result(ScannerBucketScanScope::default()); + }; + let capability_acknowledgements = remote_resolution + .remote_dirty_usage_acknowledgements + .clone() + .into_iter() + .map(Into::into) + .collect::>(); + matches!( + notification_system + .scanner_scoped_dirty_usage_capabilities(capability_acknowledgements) + .await, + Ok(true) + ) + }; + if !capability_supported { return default_result(ScannerBucketScanScope::default()); } } - return ScannerBucketScopeResolutionResult { - scope, - remote_dirty_usage_acknowledgements: scoped_acknowledgements, - }; + return remote_resolution; } default_result(scoped_scan_scope_from_dirty_buckets( @@ -245,6 +250,18 @@ where )) } +#[cfg(test)] +pub(super) async fn resolve_scanner_bucket_scan_scope_for_tests( + store: &S, + distributed: bool, + resolution: ScannerBucketScopeResolution<'_>, +) -> ScannerBucketScopeResolutionResult +where + S: ScannerStorage, +{ + resolve_scanner_bucket_scan_scope(store, distributed, resolution).await +} + pub(crate) async fn nsscanner_with_storage_status_scoped(store: &S, request: ScannerCycleRequest) -> Result where S: ScannerStorage, @@ -382,6 +399,10 @@ where dirty_usage_snapshot: &dirty_usage_snapshot, all_buckets: &all_buckets, requires_full_scan: requires_full_scan || scan_mode == HealScanMode::Deep, + #[cfg(test)] + test_peer_snapshots: None, + #[cfg(test)] + test_scoped_dirty_usage_capability: None, }, ) .await; diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index cd6a8cbd0..c45ca1f7e 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -37,6 +37,7 @@ use rustfs_concurrency::{ }; use rustfs_filemeta::FileInfo; use serial_test::serial; +use std::collections::BTreeMap; use std::sync::Arc; use temp_env::with_var; use time::OffsetDateTime; @@ -1770,6 +1771,56 @@ fn scanner_scoped_dirty_usage_ack_cost_threshold_is_single_protocol_batch() { )])); } +#[test] +fn remote_dirty_usage_scope_resolution_falls_back_when_ack_batch_exceeds_threshold() { + let source = DataUsageCacheSource::new(1, 2); + let expected_sources = HashSet::from([source]); + let scan_plan_digest = DataUsageScanPlanDigest([7; 32]); + let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11); + let bucket_names = (0..=crate::SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES) + .map(|index| format!("remote-{index:02}")) + .collect::>(); + let bucket_refs = bucket_names.iter().map(|bucket| (bucket.as_str(), 7)).collect::>(); + let all_buckets = bucket_names.iter().map(|bucket| bucket_info(bucket)).collect::>(); + let expected_peers = HashMap::from([( + "node-a:9000".to_string(), + ScannerPeerDirtyUsageExpectation { + instance_id: "instance-a".to_string(), + generation: 7, + pending: true, + }, + )]); + let remote_dirty_usage = verified_remote_dirty_usage( + &expected_peers, + vec![("node-a:9000".to_string(), peer_dirty_usage_snapshot("instance-a", 7, true, &bucket_refs))], + ) + .expect("fixture peer state should verify before the resolver cost gate"); + + let result = resolve_remote_dirty_usage_scope( + ScannerBucketScanScope::default(), + HashSet::new(), + remote_dirty_usage, + &all_buckets, + ScannerCacheBaselineProof { + authoritative_data: Some(&baseline), + observed_candidate_data: None, + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 8, + scan_plan_digest, + }, + ); + + assert!( + result.scope.is_default(), + "oversized scoped ACK batches must force the production resolver back to a full scan" + ); + assert!( + result.remote_dirty_usage_acknowledgements.is_empty(), + "full-scan fallback must not send a scoped ACK that peers would reject or split" + ); +} + #[test] fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state() { let expected_peers = HashMap::from([( @@ -1794,6 +1845,124 @@ fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state() } } +#[tokio::test] +#[serial] +async fn distributed_scoped_scan_falls_back_when_remote_ack_exceeds_protocol_batch() { + let (_temp_dir, store) = setup_two_pool_scanner_store().await; + clear_dirty_usage_buckets_for_tests(); + record_dirty_usage_bucket("local-dirty"); + let local_generation = dirty_usage_generation(); + let remote_dirty_buckets = (0..=crate::SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES) + .map(|index| (format!("remote-{index:02}"), 7)) + .collect::>(); + let mut all_buckets = vec![bucket_info_with_created_time("local-dirty")]; + all_buckets.extend( + remote_dirty_buckets + .iter() + .map(|(bucket, _)| bucket_info_with_created_time(bucket)), + ); + let snapshot_buckets = remote_dirty_buckets + .iter() + .map(|(bucket, generation)| (bucket.as_str(), *generation)) + .collect::>(); + let baseline_digest = DataUsageScanPlanDigest([8; 32]); + let baseline = complete_usage_baseline(DataUsageCacheSource::new(1, 2), baseline_digest, 7, 11); + let expected_sources = HashSet::from([DataUsageCacheSource::new(1, 2)]); + let dirty_usage_snapshot = snapshot_dirty_usage_buckets(&all_buckets, local_generation); + let activity_before = BTreeMap::from([( + "node-a:9000".to_string(), + crate::scanner::scanner_node_activity_for_tests("instance-a", 5, 7, true), + )]); + + let result = super::io_cycle::resolve_scanner_bucket_scan_scope_for_tests( + store.as_ref(), + true, + super::io_cycle::ScannerBucketScopeResolution { + requested_scope: ScannerBucketScanScope::default(), + baseline_proof: ScannerCacheBaselineProof { + authoritative_data: Some(&baseline), + observed_candidate_data: None, + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 8, + scan_plan_digest: baseline_digest, + }, + activity_before: &activity_before, + dirty_usage_snapshot: &dirty_usage_snapshot, + all_buckets: &all_buckets, + requires_full_scan: false, + test_peer_snapshots: Some(vec![( + "node-a:9000".to_string(), + peer_dirty_usage_snapshot("instance-a", 7, true, &snapshot_buckets), + )]), + test_scoped_dirty_usage_capability: Some(true), + }, + ) + .await; + + assert!( + result.scope.is_default(), + "remote scoped acknowledgements above one protocol batch must force a full scan" + ); + assert!( + result.remote_dirty_usage_acknowledgements.is_empty(), + "full-scan fallback must not send scoped remote acknowledgements" + ); + clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +async fn distributed_scoped_scan_falls_back_when_remote_scoped_ack_capability_is_rejected() { + let (_temp_dir, store) = setup_two_pool_scanner_store().await; + let source = DataUsageCacheSource::new(1, 2); + let expected_sources = HashSet::from([source]); + let scan_plan_digest = DataUsageScanPlanDigest([7; 32]); + let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11); + let dirty_usage_snapshot = DirtyUsageSnapshot { + buckets: Arc::new(HashMap::new()), + scopes: Arc::new(HashMap::new()), + generation: 7, + covers_all_pending: true, + }; + let activity_before = BTreeMap::from([( + "node-a:9000".to_string(), + crate::scanner::scanner_node_activity_for_tests("instance-a", 5, 7, true), + )]); + + for (capability, expected_buckets, expected_ack_count) in + [(true, Some(HashSet::from(["photos".to_string()])), 1), (false, None, 0)] + { + let result = super::io_cycle::resolve_scanner_bucket_scan_scope_for_tests( + store.as_ref(), + true, + super::io_cycle::ScannerBucketScopeResolution { + requested_scope: ScannerBucketScanScope::default(), + baseline_proof: ScannerCacheBaselineProof { + authoritative_data: Some(&baseline), + observed_candidate_data: None, + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 8, + scan_plan_digest, + }, + activity_before: &activity_before, + dirty_usage_snapshot: &dirty_usage_snapshot, + all_buckets: &[bucket_info_with_created_time("photos")], + requires_full_scan: false, + test_peer_snapshots: Some(vec![( + "node-a:9000".to_string(), + peer_dirty_usage_snapshot("instance-a", 7, true, &[("photos", 7)]), + )]), + test_scoped_dirty_usage_capability: Some(capability), + }, + ) + .await; + + assert_eq!(result.scope.selected_buckets.as_deref(), expected_buckets.as_ref()); + assert_eq!(result.remote_dirty_usage_acknowledgements.len(), expected_ack_count); + } +} + fn bucket_info_with_created_time(name: &str) -> BucketInfo { BucketInfo { created: Some(time::OffsetDateTime::UNIX_EPOCH),