From 3d46ed312abd0febb07de0244b9407bd52cfb93f Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Sun, 6 Sep 2026 17:52:52 +0800 Subject: [PATCH] feat(scanner): reuse complete observed scan candidates (#7206) --- crates/scanner/src/scanner.rs | 20 +++ crates/scanner/src/scanner_io.rs | 53 ++++++-- crates/scanner/src/scanner_io/io_cycle.rs | 6 +- crates/scanner/src/scanner_io/tests.rs | 123 +++++++++++++++++- .../scanner_io/tests/scoped_entry_fallback.rs | 1 + .../src/scanner_io/tests/service_cohort.rs | 1 + 6 files changed, 185 insertions(+), 19 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index dd6d00bf6..65f76ac2a 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1730,6 +1730,7 @@ where }; let baseline_publication_epoch = baseline_publication_guard.epoch(); let usage_persist_baseline_result = read_data_usage_persist_baseline(storeapi.clone()).await; + let observed_usage_candidate_result = read_config(storeapi.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()).await; drop(baseline_publication_guard); let usage_persist_baseline = match usage_persist_baseline_result { Ok(baseline) => baseline, @@ -1750,6 +1751,24 @@ where return ScannerCycleOutcome::Failed; } }; + let observed_usage_candidate = match observed_usage_candidate_result { + Ok(candidate) => Some(Bytes::from(candidate)), + Err(EcstoreError::ConfigNotFound) => None, + Err(err) => { + debug!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + cycle = cycle_info.current, + path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + state = "observed_candidate_load_failed", + error = %err, + "Scanner skipped an unavailable observed usage candidate for scoped refresh" + ); + None + } + }; let (sender, receiver) = mpsc::channel::(1); let done_cycle = Metrics::time(Metric::ScanCycle); @@ -1764,6 +1783,7 @@ where scan_mode, scan_scope: crate::scanner_io::ScannerBucketScanScope::default(), persisted_usage_baseline: usage_persist_baseline.data.clone(), + observed_usage_candidate, requires_full_scan: scheduling.requires_full_scan, service_cohort: scheduling.service_cohort, #[cfg(test)] diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index b0bac1740..3ec8cb5ab 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -27,7 +27,7 @@ use metrics::counter; use rand::seq::SliceRandom as _; #[cfg(test)] use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS}; -use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo}; +use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo, observed_data_usage_is_newer}; use rustfs_filemeta::FileMeta; use rustfs_heal_contracts::heal_channel::HealScanMode; use rustfs_lock::{LockError, NamespaceLockGuard}; @@ -128,7 +128,8 @@ impl ScannerBucketScanScope { #[derive(Clone, Copy)] pub(super) struct ScannerCacheBaselineProof<'a> { - pub(super) data: Option<&'a Bytes>, + pub(super) authoritative_data: Option<&'a Bytes>, + pub(super) observed_candidate_data: Option<&'a Bytes>, pub(super) expected_sources: &'a HashSet, pub(super) leader_epoch: u64, pub(super) want_cycle: u64, @@ -171,21 +172,23 @@ fn verified_remote_dirty_usage_buckets( (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() +fn complete_scanner_cache_snapshot_plan_digest( + snapshot: &DataUsageInfo, + proof: ScannerCacheBaselineProof<'_>, + expected_converged: bool, +) -> Option { + if !snapshot.is_complete_bucket_usage_snapshot() + || snapshot.usage_snapshot_partial + || snapshot.usage_snapshot_converged != Some(expected_converged) + || snapshot.scanner_epoch != Some(proof.leader_epoch) + || snapshot.usage_snapshot_set_states.len() != proof.expected_sources.len() { return None; } // Completed maintenance also covers ordinary usage. Keep its exact stored // proof for cache reuse, and reject mixtures of different set work proofs. - let baseline_plan_digest = DataUsageScanPlanDigest(baseline.usage_snapshot_set_states.first()?.scan_plan_digest?); + let baseline_plan_digest = DataUsageScanPlanDigest(snapshot.usage_snapshot_set_states.first()?.scan_plan_digest?); if ![ proof.scan_plan_digest, scanner_bucket_work_digest(proof.scan_plan_digest, HealScanMode::Normal, true), @@ -195,8 +198,8 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof< { return None; } - let mut states = HashSet::with_capacity(baseline.usage_snapshot_set_states.len()); - for state in &baseline.usage_snapshot_set_states { + let mut states = HashSet::with_capacity(snapshot.usage_snapshot_set_states.len()); + for state in &snapshot.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) @@ -213,6 +216,30 @@ fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof< (states == *proof.expected_sources).then_some(baseline_plan_digest) } +fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option { + let authoritative = serde_json::from_slice::(proof.authoritative_data?).ok()?; + if complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true).is_some() { + return Some(proof.scan_plan_digest); + } + + // A complete but superseded observation may reuse its per-set cache only + // when it was explicitly tied to the durable authoritative baseline. It + // remains observational: this proof grants bucket-scope reuse only and + // never changes authoritative usage publication or dirty acknowledgement. + let authoritative_has_identity = (crate::scanner::data_usage_info_has_persisted_baseline_identity(&authoritative) + && authoritative.usage_snapshot_converged != Some(false)) + || crate::scanner::data_usage_info_is_bootstrap_pending(&authoritative); + if !authoritative_has_identity { + return None; + } + let observed = serde_json::from_slice::(proof.observed_candidate_data?).ok()?; + if !observed_data_usage_is_newer(&observed, &authoritative) { + return None; + } + + complete_scanner_cache_snapshot_plan_digest(&observed, proof, false) +} + fn scoped_scan_scope_from_dirty_buckets( requested_scope: ScannerBucketScanScope, dirty_buckets: HashSet, diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 8cf9a7ffe..4eefe2910 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -72,6 +72,7 @@ where scan_mode, scan_scope: ScannerBucketScanScope::default(), persisted_usage_baseline: None, + observed_usage_candidate: None, requires_full_scan: true, service_cohort: None, #[cfg(test)] @@ -89,6 +90,7 @@ pub(crate) struct ScannerCycleRequest { pub(crate) scan_mode: HealScanMode, pub(crate) scan_scope: ScannerBucketScanScope, pub(crate) persisted_usage_baseline: Option, + pub(crate) observed_usage_candidate: Option, /// Scheduled maintenance must visit clean buckets even with a valid dirty scope. pub(crate) requires_full_scan: bool, pub(crate) service_cohort: Option>>, @@ -185,6 +187,7 @@ where scan_mode, scan_scope, persisted_usage_baseline, + observed_usage_candidate, requires_full_scan, service_cohort, #[cfg(test)] @@ -297,7 +300,8 @@ where ScannerBucketScopeResolution { requested_scope: scan_scope, baseline_proof: ScannerCacheBaselineProof { - data: persisted_usage_baseline.as_ref(), + authoritative_data: persisted_usage_baseline.as_ref(), + observed_candidate_data: observed_usage_candidate.as_ref(), expected_sources: &expected_sources, leader_epoch, want_cycle, diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 58811e947..e19fc2496 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -397,6 +397,7 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work() scan_mode, scan_scope: requested_scope, persisted_usage_baseline: baseline, + observed_usage_candidate: None, requires_full_scan, resolved_scope_observer: Some(observer), service_cohort: None, @@ -470,6 +471,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure( scan_mode: HealScanMode::Normal, scan_scope: ScannerBucketScanScope::default(), persisted_usage_baseline: None, + observed_usage_candidate: None, requires_full_scan: false, resolved_scope_observer: None, service_cohort: None, @@ -521,6 +523,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure( scan_mode, scan_scope: ScannerBucketScanScope::default(), persisted_usage_baseline: None, + observed_usage_candidate: None, requires_full_scan, resolved_scope_observer: None, service_cohort: None, @@ -1331,7 +1334,8 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance( assert_eq!( complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof { - data: Some(&baseline), + authoritative_data: Some(&baseline), + observed_candidate_data: None, expected_sources: &expected_sources, leader_epoch: 11, want_cycle: 8, @@ -1345,7 +1349,8 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance( 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), + authoritative_data: Some(&incomplete), + observed_candidate_data: None, expected_sources: &expected_sources, leader_epoch: 11, want_cycle: 8, @@ -1359,7 +1364,8 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance( 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), + authoritative_data: Some(&wrong_provenance), + observed_candidate_data: None, expected_sources: &expected_sources, leader_epoch: 11, want_cycle: 8, @@ -1369,6 +1375,111 @@ fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance( ); } +#[test] +fn scoped_scan_accepts_only_a_complete_observation_tied_to_the_authoritative_baseline() { + let source = DataUsageCacheSource::new(1, 2); + let expected_sources = HashSet::from([source]); + let scan_plan_digest = DataUsageScanPlanDigest([9; 32]); + let authoritative = complete_usage_baseline(source, scan_plan_digest, 7, 11); + let authoritative_info = + serde_json::from_slice::(&authoritative).expect("authoritative baseline should decode"); + let bootstrap_authoritative_info = + crate::scanner::scanner_usage_bootstrap_marker(SystemTime::UNIX_EPOCH + Duration::from_secs(9), Some(11)); + let bootstrap_authoritative = + bytes::Bytes::from(serde_json::to_vec(&bootstrap_authoritative_info).expect("bootstrap baseline should encode")); + let mut observed_info = authoritative_info.clone(); + observed_info.last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(11)); + observed_info.scanner_cycle = Some(8); + observed_info.usage_snapshot_converged = Some(false); + observed_info.usage_snapshot_authoritative_baseline = Some(bootstrap_authoritative_info.snapshot_identity()); + observed_info.usage_snapshot_set_states[0].scanner_cycle = Some(8); + let observed = bytes::Bytes::from(serde_json::to_vec(&observed_info).expect("observation should encode")); + + macro_rules! proof { + ($authoritative:expr, $candidate:expr) => { + ScannerCacheBaselineProof { + authoritative_data: Some($authoritative), + observed_candidate_data: $candidate, + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 9, + scan_plan_digest, + } + }; + } + assert_eq!( + complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&observed))), + Some(scan_plan_digest) + ); + + observed_info.usage_snapshot_authoritative_baseline = Some(DataUsageInfo::default().snapshot_identity()); + let mismatched_baseline = bytes::Bytes::from(serde_json::to_vec(&observed_info).expect("observation should encode")); + assert_eq!( + complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&mismatched_baseline))), + None + ); + + observed_info = serde_json::from_slice(&observed).expect("observation should decode"); + observed_info.usage_snapshot_partial = true; + let partial = bytes::Bytes::from(serde_json::to_vec(&observed_info).expect("partial observation should encode")); + assert_eq!( + complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&partial))), + None + ); + + observed_info = serde_json::from_slice(&observed).expect("observation should decode"); + observed_info.usage_snapshot_converged = Some(true); + let converged = bytes::Bytes::from(serde_json::to_vec(&observed_info).expect("converged observation should encode")); + assert_eq!( + complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&converged))), + None + ); + + let mut legacy_authoritative = authoritative_info.clone(); + legacy_authoritative.usage_snapshot_converged = None; + let mut stale_info = serde_json::from_slice::(&observed).expect("observation should decode"); + stale_info.scanner_cycle = Some(7); + stale_info.usage_snapshot_set_states[0].scanner_cycle = Some(7); + stale_info.usage_snapshot_authoritative_baseline = Some(legacy_authoritative.snapshot_identity()); + let legacy_authoritative = + bytes::Bytes::from(serde_json::to_vec(&legacy_authoritative).expect("legacy baseline should encode")); + let stale = bytes::Bytes::from(serde_json::to_vec(&stale_info).expect("stale observation should encode")); + assert_eq!( + complete_scanner_cache_baseline_plan_digest(proof!(&legacy_authoritative, Some(&stale))), + None + ); + + let malformed = bytes::Bytes::from_static(b"not data usage json"); + assert_eq!( + complete_scanner_cache_baseline_plan_digest(proof!(&bootstrap_authoritative, Some(&malformed))), + None + ); + + let mut nonconverged_authoritative = authoritative_info; + nonconverged_authoritative.usage_snapshot_converged = Some(false); + let mut observation_of_nonconverged_authoritative = nonconverged_authoritative.clone(); + observation_of_nonconverged_authoritative.last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(12)); + observation_of_nonconverged_authoritative.scanner_cycle = Some(8); + observation_of_nonconverged_authoritative.usage_snapshot_set_states[0].scanner_cycle = Some(8); + observation_of_nonconverged_authoritative.usage_snapshot_authoritative_baseline = + Some(nonconverged_authoritative.snapshot_identity()); + let nonconverged_authoritative = + bytes::Bytes::from(serde_json::to_vec(&nonconverged_authoritative).expect("nonconverged baseline should encode")); + let observation_of_nonconverged_authoritative = + bytes::Bytes::from(serde_json::to_vec(&observation_of_nonconverged_authoritative).expect("observation should encode")); + assert_eq!( + complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof { + authoritative_data: Some(&nonconverged_authoritative), + observed_candidate_data: Some(&observation_of_nonconverged_authoritative), + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 9, + scan_plan_digest, + }), + None + ); +} + #[test] fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() { let source = DataUsageCacheSource::new(1, 2); @@ -1382,7 +1493,8 @@ fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() { true, &[bucket_info("photos")], ScannerCacheBaselineProof { - data: Some(&baseline), + authoritative_data: Some(&baseline), + observed_candidate_data: None, expected_sources: &expected_sources, leader_epoch: 11, want_cycle: 8, @@ -1420,7 +1532,8 @@ fn scoped_scan_baseline_work_proof_requires_uniform_known_set_identity() { let data = Bytes::from(serde_json::to_vec(&candidate).expect("candidate should encode")); assert_eq!( complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof { - data: Some(&data), + authoritative_data: Some(&data), + observed_candidate_data: None, expected_sources: &sources, leader_epoch: 11, want_cycle: 8, diff --git a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs index ae04cc03c..31de76153 100644 --- a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs +++ b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs @@ -126,6 +126,7 @@ async fn run_entry(store: &Arc, cycle: u64, selected: Option<&str>, exp scan_mode: HealScanMode::Normal, scan_scope: ScannerBucketScanScope::default(), persisted_usage_baseline: root_before.0.clone().map(Bytes::from), + observed_usage_candidate: None, requires_full_scan: false, service_cohort: None, resolved_scope_observer: Some(observer), diff --git a/crates/scanner/src/scanner_io/tests/service_cohort.rs b/crates/scanner/src/scanner_io/tests/service_cohort.rs index a2d5ffe3a..c720fb8e1 100644 --- a/crates/scanner/src/scanner_io/tests/service_cohort.rs +++ b/crates/scanner/src/scanner_io/tests/service_cohort.rs @@ -61,6 +61,7 @@ async fn run_cohort_cycle( scan_mode: HealScanMode::Normal, scan_scope: ScannerBucketScanScope::default(), persisted_usage_baseline: None, + observed_usage_candidate: None, requires_full_scan: false, service_cohort: Some(cohort), resolved_scope_observer: None,