From b3a02f305e7f6dd60ada3db0db2b9a1d5e7ebe65 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 9 Sep 2026 00:48:45 +0800 Subject: [PATCH] feat(scanner): surface cold segment reuse oracle (#7536) Co-authored-by: zhi22915 --- crates/scanner/src/scanner_io.rs | 23 ++++++++++++++++- crates/scanner/src/scanner_io/io_cache.rs | 7 ++++++ crates/scanner/src/scanner_io/io_cycle.rs | 23 +++++++++++++---- crates/scanner/src/scanner_io/tests.rs | 25 +++++++++++++++++-- .../scanner_io/tests/scoped_entry_fallback.rs | 15 +++++++++++ 5 files changed, 85 insertions(+), 8 deletions(-) diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 6c357e3b5..ad308a645 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -516,6 +516,7 @@ fn scanner_segment_reuse_activation_preflight_from_proof( fn scanner_segment_reuse_activation_preflight_for_cycle( dirty_usage_snapshot: &DirtyUsageSnapshot, + distributed: bool, distributed_segment_invalidation_evidence: Option, cold_zero_walk_oracle: bool, ) -> ScannerSegmentReuseActivationPreflight { @@ -528,7 +529,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle( && dirty_usage_snapshot.generation != u64::MAX, overflow_absent: dirty_usage_snapshot.covers_all_pending, cold_zero_walk_oracle, - distributed_peer_invalidation: distributed_segment_invalidation_evidence.is_some(), + distributed_peer_invalidation: !distributed || distributed_segment_invalidation_evidence.is_some(), }) } @@ -595,6 +596,7 @@ pub struct ScannerBucketScanPlan { bucket_failures: ScannerBucketFailureState, pending_maintenance_work: Arc, cache_cycle_floor: Arc, + cold_zero_walk_reuse_observed: Arc, } #[derive(Clone, Default)] @@ -727,6 +729,25 @@ fn scanner_bucket_scan_status(has_failed: bool, has_partial: bool, has_namespace } } +fn scanner_cycle_cold_zero_walk_oracle( + scan_scope: &ScannerBucketScanScope, + all_buckets: &[BucketInfo], + completed_all_sets: bool, + scan_scope_matches: bool, + bucket_scan_status: ScannerBucketScanStatus, + cold_zero_walk_reuse_observed: bool, +) -> bool { + let Some(selected_buckets) = scan_scope.selected_buckets.as_deref() else { + return false; + }; + cold_zero_walk_reuse_observed + && !selected_buckets.is_empty() + && completed_all_sets + && scan_scope_matches + && bucket_scan_status == ScannerBucketScanStatus::Complete + && all_buckets.iter().any(|bucket| !selected_buckets.contains(&bucket.name)) +} + fn classify_nsscanner_cycle( completed_all_sets: bool, budget_elapsed: bool, diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index 2cb970d8e..c0371f85c 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -172,6 +172,7 @@ impl ScannerIOCache for SetDisks { bucket_failures, pending_maintenance_work, cache_cycle_floor, + cold_zero_walk_reuse_observed, } = scan_plan; let scan_plan_digest = scanner_bucket_work_digest(scan_plan_digest, scan_mode, requires_full_scan); let bucket_work_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan); @@ -219,6 +220,9 @@ impl ScannerIOCache for SetDisks { }, current_bucket_incarnations.as_ref(), ); + let cold_zero_walk_reuse_candidate = scoped_scan.as_ref().is_some_and(|prepared| { + old_cache.info.next_cycle < want_cycle && !prepared.buckets.is_empty() && prepared.buckets.len() < all_buckets.len() + }); let mut scoped_cache = scoped_scan.map(|mut prepared| { buckets = prepared.buckets; prepared.cache.info.scan_coverage_digest = Some(bucket_coverage_digest); @@ -1466,6 +1470,9 @@ impl ScannerIOCache for SetDisks { cache.info.lkg_last_update = None; cache.info.lkg_leader_epoch = None; cache.info.lkg_scan_plan_digest = None; + if cold_zero_walk_reuse_candidate { + cold_zero_walk_reuse_observed.store(true, Ordering::Release); + } cache.clone() }; let _ = persist_and_publish_cache_snapshot( diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 9131d9ccf..d598fff27 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -410,11 +410,6 @@ where .await; let remote_dirty_usage_acknowledgements = scope_resolution.remote_dirty_usage_acknowledgements; let distributed_segment_invalidation_evidence = scope_resolution.distributed_segment_invalidation_evidence; - let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle( - &dirty_usage_snapshot, - distributed_segment_invalidation_evidence, - false, - ); let scan_scope = scope_resolution.scope; #[cfg(test)] if let Some(observer) = resolved_scope_observer { @@ -472,6 +467,8 @@ where } else { Vec::new() }; + let segment_reuse_activation_preflight = + scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, distributed, None, false); return Ok(ScannerCycleResult::new(status, dirty_usage_clear) .with_publication_epoch(publication_epoch) .with_activity_digest(activity_digest) @@ -503,6 +500,7 @@ where ); let bucket_failures = ScannerBucketFailureState::default(); let pending_maintenance_work = Arc::new(AtomicBool::new(false)); + let cold_zero_walk_reuse_observed = Arc::new(AtomicBool::new(false)); record_set_scan_concurrency_limit(set_scan_limit); debug!( target: "rustfs::scanner::io", @@ -596,6 +594,7 @@ where bucket_failures: bucket_failures.clone(), pending_maintenance_work: pending_maintenance_work.clone(), cache_cycle_floor: cache_cycle_floor.clone(), + cold_zero_walk_reuse_observed: cold_zero_walk_reuse_observed.clone(), }; // Spawn task to run the scanner let scanner_fut = tokio::spawn(async move { @@ -699,6 +698,20 @@ where scan_scope_matches && !partial_buckets.is_empty(), scan_scope_matches && !namespace_not_found_buckets.is_empty(), ); + let cold_zero_walk_oracle = scanner_cycle_cold_zero_walk_oracle( + &scan_scope, + &all_buckets, + completed_all_sets, + scan_scope_matches, + bucket_scan_status, + cold_zero_walk_reuse_observed.load(Ordering::Acquire), + ); + let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle( + &dirty_usage_snapshot, + distributed, + distributed_segment_invalidation_evidence, + cold_zero_walk_oracle, + ); let pending_maintenance_work = pending_maintenance_work_for_cycle(&pending_maintenance_work, &results); let observed_cycle_floor = cache_cycle_floor.load(Ordering::Acquire); let required_cycle_floor = (observed_cycle_floor > want_cycle).then_some(observed_cycle_floor); diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 37dc15f03..076a1b1e6 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -166,7 +166,8 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_reports_cycle_inputs_wit all_peers_bound_to_generation_window: true, }; - let preflight = scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, Some(distributed_evidence), true); + let preflight = + scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, true, Some(distributed_evidence), true); assert!(!preflight.production_activation); assert!(!preflight.scanner_segment_reuse_activated); @@ -185,7 +186,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_blocks_unbounded_inputs( covers_all_pending: false, }; - let preflight = scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, None, false); + let preflight = scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, true, None, false); assert!(!preflight.production_activation); assert!(!preflight.scanner_segment_reuse_activated); @@ -195,6 +196,25 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_blocks_unbounded_inputs( ); } +#[test] +fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocker_for_local_scan() { + let dirty_usage_snapshot = DirtyUsageSnapshot { + buckets: Arc::new(DirtyUsageBuckets::from([("photos".to_string(), 7)])), + scopes: Arc::new(DirtyUsageBucketScopes::default()), + generation: 7, + covers_all_pending: true, + }; + + let preflight = scanner_segment_reuse_activation_preflight_for_cycle(&dirty_usage_snapshot, false, None, true); + + assert!(!preflight.production_activation); + assert!(!preflight.scanner_segment_reuse_activated); + assert_eq!( + preflight.fail_closed_blockers().collect::>(), + vec!["missing_producer_identity", "restart_gap"] + ); +} + #[test] fn scanner_cycle_result_returns_segment_reuse_activation_preflight() { let proof = ScannerSegmentReuseActivationProof { @@ -1501,6 +1521,7 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers bucket_failures: ScannerBucketFailureState::default(), pending_maintenance_work: Arc::new(AtomicBool::new(false)), cache_cycle_floor: Arc::new(AtomicU64::new(8)), + cold_zero_walk_reuse_observed: Arc::new(AtomicBool::new(false)), }, tx, 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 c6bc58c3e..416677c57 100644 --- a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs +++ b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs @@ -138,6 +138,7 @@ async fn run_entry(store: &Arc, cycle: u64, selected: Option<&str>, exp .expect("entry cycle should finish within the fixture deadline") .expect("entry cycle should succeed"); assert_eq!(result.status, ScannerCycleStatus::Complete); + let activation_preflight = result.segment_reuse_activation_preflight; let scope = observed.await.expect("production resolver should report its decision"); assert_eq!( scope.selected_buckets.as_deref(), @@ -174,6 +175,20 @@ async fn run_entry(store: &Arc, cycle: u64, selected: Option<&str>, exp actual, expected_walks, "each listed source/bucket must have exactly the expected real walks" ); + assert!(!activation_preflight.production_activation); + assert!(!activation_preflight.scanner_segment_reuse_activated); + let activation_blockers = activation_preflight.fail_closed_blockers().collect::>(); + if selected.is_some() && expect_walks { + assert!( + !activation_blockers.contains(&"missing_cold_zero_walk_oracle"), + "a complete scoped reuse cycle must carry the cold zero-walk oracle: cycle={cycle} selected={selected:?} blockers={activation_blockers:?}" + ); + } else { + assert!( + activation_blockers.contains(&"missing_cold_zero_walk_oracle"), + "unscoped or same-cycle cache reuse must not claim the cold zero-walk oracle: cycle={cycle} selected={selected:?} expect_walks={expect_walks} blockers={activation_blockers:?}" + ); + } assert_eq!( read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) .await