From 444eb5ddbaa567f4dc327bb6ea11fb8388a87e7f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Wed, 21 Oct 2026 13:55:18 +0800 Subject: [PATCH] fix(scanner): admit dirty scope across the bucket-listing window and pace full-walk fallbacks (#8690) Under sustained writes nearly every cycle fell back to a full walk: scope admission required the global dirty generation to stay unchanged across the bucket listing, and after a Complete cycle the dirty-usage wake ran the next full walk immediately, back to back. - Track dirty-map removals with a process-local removal generation and admit a scoped walk when no entry was removed or replayed during the listing (`covers_listed_pending`). Writes to listed buckets in the window join the start snapshot; unlisted buckets dirtied after the cutoff stay pending and are never cleared by compare-and-remove certification. The strict `covers_all_pending` still gates segment-reuse evidence. - Classify scope fallbacks (removal during listing, generation exhausted, dirty-bucket overflow, no listed dirty bucket, baseline unavailable, peer scope unavailable), log `scoped_scan_fallback`, and report them in usage_freshness as additive serde(default) fields. - Pace the cycle after a Complete fallback while dirty buckets remain on the capped retry schedule clamped to the cycle interval, with the dirty wake disabled; the paced timer wake still requests a scope. - Update the per-bucket convergence design doc. No persisted format or internode protocol change. Co-authored-by: Chris --- .../ecstore/src/services/metrics_realtime.rs | 9 + crates/madmin/src/metrics.rs | 33 +++ crates/scanner-metrics/src/metrics.rs | 34 +++ crates/scanner/src/scanner.rs | 50 ++++- crates/scanner/src/scanner/activity.rs | 21 ++ crates/scanner/src/scanner/tests.rs | 80 +++++++ .../scanner/tests/cycle_budget_publication.rs | 5 + crates/scanner/src/scanner_io.rs | 101 +++++++++ crates/scanner/src/scanner_io/dirty_usage.rs | 86 +++++++- crates/scanner/src/scanner_io/io_cycle.rs | 80 ++++++- crates/scanner/src/scanner_io/tests.rs | 7 + .../scanner_io/tests/listing_window_scope.rs | 206 ++++++++++++++++++ .../scanner-per-bucket-convergence.md | 42 +++- 13 files changed, 733 insertions(+), 21 deletions(-) create mode 100644 crates/scanner/src/scanner_io/tests/listing_window_scope.rs diff --git a/crates/ecstore/src/services/metrics_realtime.rs b/crates/ecstore/src/services/metrics_realtime.rs index 551bae35c..0bda1559d 100644 --- a/crates/ecstore/src/services/metrics_realtime.rs +++ b/crates/ecstore/src/services/metrics_realtime.rs @@ -239,6 +239,9 @@ fn to_madmin_scanner_metrics(metrics: rustfs_scanner_metrics::metrics::ScannerMe guardrail_veto_total: metrics.usage_freshness.guardrail_veto_total, last_guardrail_veto_unix_secs: metrics.usage_freshness.last_guardrail_veto_unix_secs, last_guardrail_veto_reason: metrics.usage_freshness.last_guardrail_veto_reason, + scope_fallback_total: metrics.usage_freshness.scope_fallback_total, + last_scope_fallback_unix_secs: metrics.usage_freshness.last_scope_fallback_unix_secs, + last_scope_fallback_reason: metrics.usage_freshness.last_scope_fallback_reason, disk_io_budget_wait_events: metrics.usage_freshness.disk_io_budget_wait_events, disk_io_budget_wait_millis: metrics.usage_freshness.disk_io_budget_wait_millis, last_disk_io_budget_wait_unix_secs: metrics.usage_freshness.last_disk_io_budget_wait_unix_secs, @@ -901,6 +904,9 @@ mod test { guardrail_veto_total: 4, last_guardrail_veto_unix_secs: 17, last_guardrail_veto_reason: "publication_blocked@".to_string(), + scope_fallback_total: 5, + last_scope_fallback_unix_secs: 18, + last_scope_fallback_reason: "dirty_snapshot_incomplete".to_string(), disk_io_budget_wait_events: 5, disk_io_budget_wait_millis: 2_500, last_disk_io_budget_wait_unix_secs: 18, @@ -910,6 +916,9 @@ mod test { assert_eq!(scanner.usage_freshness.guardrail_veto_total, 4); assert_eq!(scanner.usage_freshness.last_guardrail_veto_unix_secs, 17); + assert_eq!(scanner.usage_freshness.scope_fallback_total, 5); + assert_eq!(scanner.usage_freshness.last_scope_fallback_unix_secs, 18); + assert_eq!(scanner.usage_freshness.last_scope_fallback_reason, "dirty_snapshot_incomplete"); assert_eq!(scanner.usage_freshness.last_guardrail_veto_reason, "publication_blocked@"); assert_eq!(scanner.usage_freshness.dirty_pending_buckets, 3); assert_eq!(scanner.usage_freshness.last_dirty_mark_unix_secs, 10); diff --git a/crates/madmin/src/metrics.rs b/crates/madmin/src/metrics.rs index c9b555338..4b4b9b059 100644 --- a/crates/madmin/src/metrics.rs +++ b/crates/madmin/src/metrics.rs @@ -325,6 +325,14 @@ pub struct ScannerUsageFreshnessSnapshot { pub last_guardrail_veto_unix_secs: u64, #[serde(rename = "last_guardrail_veto_reason", default)] pub last_guardrail_veto_reason: String, + /// Cycles eligible for a dirty-bucket scope that walked the whole + /// namespace instead; the reason names the scope precondition that failed. + #[serde(rename = "scope_fallback_total", default)] + pub scope_fallback_total: u64, + #[serde(rename = "last_scope_fallback_unix_secs", default)] + pub last_scope_fallback_unix_secs: u64, + #[serde(rename = "last_scope_fallback_reason", default)] + pub last_scope_fallback_reason: String, #[serde(rename = "disk_io_budget_wait_events", default)] pub disk_io_budget_wait_events: u64, #[serde(rename = "disk_io_budget_wait_millis", default)] @@ -370,6 +378,11 @@ impl ScannerUsageFreshnessSnapshot { self.last_guardrail_veto_unix_secs = other.last_guardrail_veto_unix_secs; self.last_guardrail_veto_reason = other.last_guardrail_veto_reason.clone(); } + self.scope_fallback_total = self.scope_fallback_total.saturating_add(other.scope_fallback_total); + if other.last_scope_fallback_unix_secs > self.last_scope_fallback_unix_secs { + self.last_scope_fallback_unix_secs = other.last_scope_fallback_unix_secs; + self.last_scope_fallback_reason = other.last_scope_fallback_reason.clone(); + } if other_deferred_state_at > self_deferred_state_at { self.deferred_pending = other.deferred_pending; } else if other_deferred_state_at == self_deferred_state_at { @@ -1773,6 +1786,26 @@ mod tests { assert_eq!(legacy.guardrail_veto_total, 0); assert_eq!(legacy.last_guardrail_veto_unix_secs, 0); assert!(legacy.last_guardrail_veto_reason.is_empty()); + assert_eq!(legacy.scope_fallback_total, 0); + assert_eq!(legacy.last_scope_fallback_unix_secs, 0); + assert!(legacy.last_scope_fallback_reason.is_empty()); + + let mut fallbacks = ScannerUsageFreshnessSnapshot::default(); + fallbacks.merge(&ScannerUsageFreshnessSnapshot { + scope_fallback_total: 3, + last_scope_fallback_unix_secs: 40, + last_scope_fallback_reason: "dirty_bucket_overflow".to_string(), + ..Default::default() + }); + fallbacks.merge(&ScannerUsageFreshnessSnapshot { + scope_fallback_total: 1, + last_scope_fallback_unix_secs: 25, + last_scope_fallback_reason: "baseline_unavailable".to_string(), + ..Default::default() + }); + assert_eq!(fallbacks.scope_fallback_total, 4); + assert_eq!(fallbacks.last_scope_fallback_unix_secs, 40); + assert_eq!(fallbacks.last_scope_fallback_reason, "dirty_bucket_overflow"); let mut vetoes = ScannerUsageFreshnessSnapshot::default(); vetoes.merge(&ScannerUsageFreshnessSnapshot { diff --git a/crates/scanner-metrics/src/metrics.rs b/crates/scanner-metrics/src/metrics.rs index cecebf903..13be891c1 100644 --- a/crates/scanner-metrics/src/metrics.rs +++ b/crates/scanner-metrics/src/metrics.rs @@ -1042,6 +1042,9 @@ pub struct Metrics { scanner_guardrail_veto_total: AtomicU64, scanner_guardrail_last_veto_unix_secs: AtomicU64, scanner_guardrail_last_veto_reason: Mutex, + scanner_scope_fallback_total: AtomicU64, + scanner_scope_fallback_last_unix_secs: AtomicU64, + scanner_scope_fallback_last_reason: Mutex, scanner_usage_last_save_result: AtomicU8, scanner_usage_deferred_pending: AtomicBool, scanner_usage_deferred_total: AtomicU64, @@ -1382,6 +1385,14 @@ pub struct ScannerUsageFreshnessSnapshot { pub last_guardrail_veto_unix_secs: u64, #[serde(default)] pub last_guardrail_veto_reason: String, + /// Cycles eligible for a dirty-bucket scope that walked the whole + /// namespace instead, with the reason of the last one. + #[serde(default)] + pub scope_fallback_total: u64, + #[serde(default)] + pub last_scope_fallback_unix_secs: u64, + #[serde(default)] + pub last_scope_fallback_reason: String, /// Waits imposed by the per-disk background I/O budget; sustained waits /// explain slow usage freshness on a budgeted disk. #[serde(default)] @@ -2162,6 +2173,9 @@ impl Metrics { scanner_guardrail_veto_total: AtomicU64::new(0), scanner_guardrail_last_veto_unix_secs: AtomicU64::new(0), scanner_guardrail_last_veto_reason: Mutex::new(String::new()), + scanner_scope_fallback_total: AtomicU64::new(0), + scanner_scope_fallback_last_unix_secs: AtomicU64::new(0), + scanner_scope_fallback_last_reason: Mutex::new(String::new()), scanner_usage_last_save_result: AtomicU8::new(ScannerUsageSaveResult::Unknown as u8), scanner_usage_deferred_pending: AtomicBool::new(false), scanner_usage_deferred_total: AtomicU64::new(0), @@ -2588,6 +2602,20 @@ impl Metrics { *last_reason = reason.into(); } + /// Record a cycle that could have been scoped to its dirty buckets but + /// walked the whole namespace, so an operator can tell why scoped + /// convergence is not happening. + pub fn record_scanner_scope_fallback(&self, reason: impl Into) { + self.scanner_scope_fallback_total.fetch_add(1, Ordering::Relaxed); + self.scanner_scope_fallback_last_unix_secs + .store(unix_now_secs(), Ordering::Relaxed); + let mut last_reason = match self.scanner_scope_fallback_last_reason.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + *last_reason = reason.into(); + } + pub fn record_scanner_usage_durable_success(&self) { self.record_scanner_usage_publication("success", ""); self.scanner_usage_last_durable_success_unix_secs @@ -3720,6 +3748,12 @@ impl Metrics { Ok(reason) => reason.clone(), Err(poisoned) => poisoned.into_inner().clone(), }, + scope_fallback_total: self.scanner_scope_fallback_total.load(Ordering::Relaxed), + last_scope_fallback_unix_secs: self.scanner_scope_fallback_last_unix_secs.load(Ordering::Relaxed), + last_scope_fallback_reason: match self.scanner_scope_fallback_last_reason.lock() { + Ok(reason) => reason.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + }, }; m.throttle_idle_mode_enabled = self.scanner_throttle_idle_mode_enabled.load(Ordering::Relaxed); m.throttle_sleep_factor = self.scanner_throttle_sleep_factor_micros.load(Ordering::Relaxed) as f64 / 1_000_000.0; diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 27cf0b372..9c276c81e 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1699,6 +1699,25 @@ async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle, cycle_metrics_guard struct ScannerCycleScheduling { requires_full_scan: bool, service_cohort: Option>>, + /// Receives the reason when a `Complete` cycle walked every bucket in + /// place of an eligible dirty-bucket scope. + scope_fallback: Option>, +} + +/// Hand-off from a cycle to the leader loop: while dirty buckets remain after +/// a full-namespace fallback, the next cycle is paced by a bounded retry +/// schedule, not by the immediate dirty-usage wake that scoped cycles use. +#[derive(Debug, Default)] +struct ScannerScopeFallbackSlot(StdMutex>); + +impl ScannerScopeFallbackSlot { + fn set(&self, reason: Option) { + *self.0.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = reason; + } + + fn take(&self) -> Option { + self.0.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).take() + } } #[cfg(test)] @@ -1723,6 +1742,7 @@ where ScannerCycleScheduling { requires_full_scan: true, service_cohort: None, + scope_fallback: None, }, ) .await @@ -1918,6 +1938,14 @@ where }, ) .await; + if let Some(slot) = &scheduling.scope_fallback { + slot.set( + scan_result + .as_ref() + .ok() + .and_then(ScannerCycleResult::completed_scope_fallback), + ); + } let publication_plan = scan_result .as_ref() .map(ScannerCyclePublicationPlan::for_result) @@ -2844,6 +2872,8 @@ where let mut clean_idle_backoff = ScannerCleanIdleBackoff::default(); let mut superseded_backoff = ScannerRetryBackoff::default(); let mut deferred_backoff = ScannerRetryBackoff::default(); + let mut scope_fallback_backoff = ScannerRetryBackoff::default(); + let scope_fallback_slot = Arc::new(ScannerScopeFallbackSlot::default()); let service_cohort = Arc::new(StdMutex::new(crate::scanner_io::ScannerServiceCohort::default())); let initial_runtime_config = resolve_scanner_runtime_config(); if clean_idle_topology_supported && maintenance_generation_seen.is_none() { @@ -3078,6 +3108,7 @@ where ScannerCycleScheduling { requires_full_scan: true, service_cohort: Some(service_cohort.clone()), + scope_fallback: None, }, ), guard.lock_lost_notified(), @@ -3225,7 +3256,10 @@ where let superseded_retry_interval = scanner_superseded_retry_interval(superseded_backoff, &runtime_config, maintenance_features); let deferred_retry_interval = deferred_backoff.retry_interval(runtime_config.cycle_interval); - let mut convergence_retry_interval = superseded_retry_interval.or(deferred_retry_interval); + let scope_fallback_retry_interval = scanner_scope_fallback_retry_interval(scope_fallback_backoff, &runtime_config); + let mut convergence_retry_interval = superseded_retry_interval + .or(scope_fallback_retry_interval) + .or(deferred_retry_interval); if let Some(retry_interval) = convergence_retry_interval { wait_plan.effective_interval = retry_interval; wait_plan.delay = randomized_cycle_delay_for(retry_interval).min(retry_interval); @@ -3265,6 +3299,8 @@ where superseded_cycles = superseded_backoff.consecutive_cycles, deferred_retry_backoff_enabled = deferred_retry_interval.is_some(), deferred_cycles = deferred_backoff.consecutive_cycles, + scope_fallback_retry_backoff_enabled = scope_fallback_retry_interval.is_some(), + scope_fallback_cycles = scope_fallback_backoff.consecutive_cycles, lifecycle_active = maintenance_features.lifecycle, replication_active = maintenance_features.replication, feature_inspection_failed = maintenance_features.inspection_failed, @@ -3404,9 +3440,10 @@ where maintenance_generation_seen, scanner_maintenance_generation(), wake_reason, - superseded_retry_interval.is_some(), + superseded_retry_interval.is_some() || scope_fallback_retry_interval.is_some(), ), service_cohort: Some(service_cohort.clone()), + scope_fallback: Some(scope_fallback_slot.clone()), }, ), guard.lock_lost_notified(), @@ -3462,6 +3499,15 @@ where outcome, ScannerCycleOutcome::Deferred(_) | ScannerCycleOutcome::StatePersistenceFailed )); + // Writes that land after the walk would otherwise wake the next full + // walk immediately; with no dirty bucket left there is nothing to pace. + let scope_fallback = scope_fallback_slot.take().filter(|_| { + matches!( + outcome, + ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance + ) && dirty_usage_buckets_pending() + }); + scope_fallback_backoff.record_retryable_cycle(scope_fallback.is_some()); dirty_usage_generation_seen = dirty_generation_before_cycle; if guard.is_lock_lost() { record_scanner_leader_lock_lost("Scanner leader lock lost during a scanner cycle").await; diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index 2375a930e..6832693af 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -167,6 +167,27 @@ pub(super) fn scanner_superseded_retry_interval( Some(retry_interval) } +/// Pacing for the cycle after a full-namespace scope fallback while dirty +/// buckets remain. The scoped cycle's immediate dirty-usage wake would run +/// such full walks back to back under sustained writes, so the next attempt +/// waits on the capped retry schedule instead. The wait never exceeds the +/// configured cycle interval: a namespace that stays unscopable is walked at +/// the ordinary cadence rather than starved by the superseded cap. The next +/// timer wake still requests a scope, so a scoped cycle resumes as soon as +/// one is admitted and resets this schedule. +pub(super) fn scanner_scope_fallback_retry_interval( + backoff: ScannerRetryBackoff, + runtime_config: &ScannerRuntimeConfig, +) -> Option { + let cycle_interval = runtime_config.cycle_interval.max(Duration::from_secs(1)); + let retry_interval = backoff.retry_interval(runtime_config.cycle_interval)?.min(cycle_interval); + // Preserve an operator-selected duty-cycle floor, as superseded retries do. + if runtime_config.cycle_interval_source != ScannerRuntimeConfigSource::Default { + return Some(cycle_interval); + } + Some(retry_interval) +} + const SCANNER_PUBLICATION_PROOF_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30); pub(crate) fn scanner_publication_proof_retry_delay(consecutive_failures: u32) -> Duration { diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 7b298aa4d..514a53629 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -1601,6 +1601,7 @@ async fn assert_bootstrap_authority_during_pending_put(drives_per_set: usize) { ScannerCycleScheduling { requires_full_scan: true, service_cohort: None, + scope_fallback: None, }, ), ) @@ -1688,6 +1689,7 @@ async fn assert_bootstrap_authority_during_pending_put(drives_per_set: usize) { true, ), service_cohort: None, + scope_fallback: None, }, ), ) @@ -8872,6 +8874,84 @@ fn superseded_retry_does_not_delay_required_maintenance() { } } +#[test] +fn scope_fallback_retry_paces_full_walks_up_to_the_cycle_interval() { + let config = ScannerRuntimeConfig { + cycle_interval: Duration::from_secs(60), + ..Default::default() + }; + let mut backoff = ScannerRetryBackoff::default(); + assert_eq!(scanner_scope_fallback_retry_interval(backoff, &config), None); + + // Repeated fallbacks back off from seconds and settle on the ordinary + // cadence: never back to back, never starved by the 30-minute cap. + for expected in [5, 10, 20, 40, 60, 60] { + backoff.record_retryable_cycle(true); + assert_eq!( + scanner_scope_fallback_retry_interval(backoff, &config), + Some(Duration::from_secs(expected)) + ); + } + for _ in 0..40 { + backoff.record_retryable_cycle(true); + } + assert_eq!(scanner_scope_fallback_retry_interval(backoff, &config), Some(config.cycle_interval)); + + // The wait disables the immediate dirty-usage wake and defers cluster + // activity, which is what keeps full walks from running back to back. + let generations = ScannerCycleObservedGenerations::for_wait( + &config, + scanner_scope_fallback_retry_interval(backoff, &config), + crate::scanner_io::dirty_usage_generation(), + crate::runtime_config::scanner_runtime_config_generation(), + crate::scanner_io::scanner_maintenance_generation(), + ); + assert!(generations.dirty_usage.is_none()); + assert!(generations.defer_cluster_activity); + // The paced timer wake still requests a scope, so a scoped cycle resumes + // as soon as one is admitted. + assert!(!ScannerMaintenanceFeatures::default().requires_full_scan( + Some(scanner_maintenance_generation()), + scanner_maintenance_generation(), + ScannerCycleWakeReason::Timer, + true, + )); + + // A scoped (or otherwise non-fallback) cycle resets the schedule. + backoff.record_retryable_cycle(false); + assert_eq!(scanner_scope_fallback_retry_interval(backoff, &config), None); +} + +#[test] +fn scope_fallback_retry_preserves_explicit_and_fast_cycle_cadence() { + let mut backoff = ScannerRetryBackoff::default(); + backoff.record_retryable_cycle(true); + for source in [ + ScannerRuntimeConfigSource::Env, + ScannerRuntimeConfigSource::Config, + ScannerRuntimeConfigSource::ScannerCompatConfig, + ] { + let config = ScannerRuntimeConfig { + cycle_interval: Duration::from_secs(6 * 60 * 60), + cycle_interval_source: source, + ..Default::default() + }; + assert_eq!( + scanner_scope_fallback_retry_interval(backoff, &config), + Some(config.cycle_interval), + "{source:?} cycle cadence is a duty-cycle floor for fallback walks" + ); + } + let fastest = ScannerRuntimeConfig { + cycle_interval: Duration::from_secs(1), + ..Default::default() + }; + for _ in 0..8 { + backoff.record_retryable_cycle(true); + } + assert_eq!(scanner_scope_fallback_retry_interval(backoff, &fastest), Some(Duration::from_secs(1))); +} + #[test] fn superseded_retry_only_caps_regular_maintenance() { let config = ScannerRuntimeConfig { diff --git a/crates/scanner/src/scanner/tests/cycle_budget_publication.rs b/crates/scanner/src/scanner/tests/cycle_budget_publication.rs index a0eb09034..102f6917a 100644 --- a/crates/scanner/src/scanner/tests/cycle_budget_publication.rs +++ b/crates/scanner/src/scanner/tests/cycle_budget_publication.rs @@ -242,6 +242,7 @@ async fn assert_coordinator_partial_progress(resume_mixed_plan: bool, fail_cycle ScannerCycleScheduling { requires_full_scan: true, service_cohort: None, + scope_fallback: None, }, ) .await; @@ -310,6 +311,7 @@ async fn assert_coordinator_partial_progress(resume_mixed_plan: bool, fail_cycle ScannerCycleScheduling { requires_full_scan: true, service_cohort: None, + scope_fallback: None, }, ), ) @@ -347,6 +349,7 @@ async fn assert_coordinator_partial_progress(resume_mixed_plan: bool, fail_cycle ScannerCycleScheduling { requires_full_scan: true, service_cohort: None, + scope_fallback: None, }, ) .await; @@ -412,6 +415,7 @@ async fn assert_coordinator_partial_progress(resume_mixed_plan: bool, fail_cycle ScannerCycleScheduling { requires_full_scan: true, service_cohort: None, + scope_fallback: None, }, ) .await; @@ -487,6 +491,7 @@ async fn assert_coordinator_partial_progress(resume_mixed_plan: bool, fail_cycle ScannerCycleScheduling { requires_full_scan: true, service_cohort: None, + scope_fallback: None, }, ) .await; diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 094ab0785..18b1f176c 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -127,7 +127,11 @@ struct DirtyUsageSnapshot { buckets: Arc, scopes: Arc, generation: u64, + /// Nothing moved in the dirty map during the bucket listing. covers_all_pending: bool, + /// Every pending bucket this cycle listed is in the snapshot; see + /// `snapshot_dirty_usage_buckets_after_listing`. Admits a scoped walk. + covers_listed_pending: bool, } #[derive(Clone, Debug, Default)] @@ -617,6 +621,80 @@ fn scoped_scan_scope_from_dirty_buckets( ScannerBucketScanScope::from_dirty_buckets(selected_buckets, selected_bucket_prefixes, baseline_scan_plan_digest) } +/// Why a cycle eligible for a dirty-bucket scope walked the whole namespace. +/// +/// Only cycles that were not required to walk everything are classified, so +/// scheduled maintenance, deep scans, and the startup cycle never report one. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum ScannerScopeFallbackReason { + /// A dirty entry was removed or replaced during the bucket listing. + DirtySnapshotIncomplete, + /// The dirty-usage generation is exhausted and cannot be verified. + DirtyGenerationExhausted, + /// More buckets are dirty than one scoped cycle may select. + DirtyBucketOverflow, + /// No dirty bucket is in the listing, so there is nothing to scope to. + NoListedDirtyBuckets, + /// No complete authoritative or observed baseline matches the plan. + BaselineUnavailable, + /// Peer dirty usage could not be verified or scoped acknowledgements + /// were rejected. + PeerScopeUnavailable, +} + +impl ScannerScopeFallbackReason { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::DirtySnapshotIncomplete => "dirty_snapshot_incomplete", + Self::DirtyGenerationExhausted => "dirty_generation_exhausted", + Self::DirtyBucketOverflow => "dirty_bucket_overflow", + Self::NoListedDirtyBuckets => "no_listed_dirty_buckets", + Self::BaselineUnavailable => "baseline_unavailable", + Self::PeerScopeUnavailable => "peer_scope_unavailable", + } + } +} + +/// Classify a resolved full-namespace scope against the preconditions +/// `resolve_scanner_bucket_scan_scope` checks, in the same order. Returns +/// `None` when the cycle was required to walk everything or was scoped. +fn scanner_scope_fallback_reason( + scope_eligible: bool, + resolved_scope: &ScannerBucketScanScope, + dirty_usage_snapshot: &DirtyUsageSnapshot, + all_buckets: &[BucketInfo], + baseline_proof: ScannerCacheBaselineProof<'_>, + distributed: bool, +) -> Option { + if !scope_eligible || !resolved_scope.is_default() { + return None; + } + if dirty_usage_snapshot.generation == u64::MAX { + return Some(ScannerScopeFallbackReason::DirtyGenerationExhausted); + } + if !dirty_usage_snapshot.covers_listed_pending { + return Some(ScannerScopeFallbackReason::DirtySnapshotIncomplete); + } + if dirty_usage_snapshot.buckets.len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES { + return Some(ScannerScopeFallbackReason::DirtyBucketOverflow); + } + // Peers may contribute dirty buckets the local snapshot does not hold. + if !distributed { + let listed = all_buckets.iter().map(|bucket| bucket.name.as_str()).collect::>(); + if !dirty_usage_snapshot + .buckets + .keys() + .any(|bucket| listed.contains(bucket.as_str())) + { + return Some(ScannerScopeFallbackReason::NoListedDirtyBuckets); + } + } + if complete_scanner_cache_baseline_plan_digest(baseline_proof).is_none() { + return Some(ScannerScopeFallbackReason::BaselineUnavailable); + } + distributed.then_some(ScannerScopeFallbackReason::PeerScopeUnavailable) +} + fn scanner_segment_reuse_activation_preflight() -> ScannerSegmentReuseActivationPreflight { scanner_segment_reuse_activation_preflight_from_proof(ScannerSegmentReuseActivationProof { production_activation: true, @@ -1652,6 +1730,7 @@ pub(crate) struct ScannerCycleResult { failed_dirty_usage: bool, pending_maintenance_work: bool, required_cycle_floor: Option, + scope_fallback: Option, publication_expectation: Option, } @@ -1673,6 +1752,7 @@ impl ScannerCycleResult { failed_dirty_usage: false, pending_maintenance_work: false, required_cycle_floor: None, + scope_fallback: None, publication_expectation: None, } } @@ -1753,6 +1833,27 @@ impl ScannerCycleResult { self } + fn with_scope_fallback(mut self, scope_fallback: Option) -> Self { + self.publication_expectation = None; + self.scope_fallback = scope_fallback; + self + } + + /// Why this cycle walked the whole namespace although a dirty-bucket + /// scope was eligible. + #[cfg(test)] + pub(crate) fn scope_fallback(&self) -> Option { + self.scope_fallback + } + + /// A `Complete` full-namespace fallback: while dirty buckets remain, the + /// next cycle must not run under the scoped cycle's immediate wake. + pub(crate) fn completed_scope_fallback(&self) -> Option { + (self.status == ScannerCycleStatus::Complete) + .then_some(self.scope_fallback) + .flatten() + } + fn with_required_cycle_floor(mut self, required_cycle_floor: Option) -> Self { self.publication_expectation = None; self.required_cycle_floor = required_cycle_floor; diff --git a/crates/scanner/src/scanner_io/dirty_usage.rs b/crates/scanner/src/scanner_io/dirty_usage.rs index ed2920142..fbf58d333 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -17,6 +17,11 @@ use std::collections::BTreeSet; use std::sync::RwLock as StdRwLock; pub(super) static DIRTY_USAGE_BUCKET_GENERATION: AtomicU64 = AtomicU64::new(0); +// Advances, under the `DIRTY_USAGE_BUCKETS` lock, whenever an entry leaves the +// dirty map or is overwritten by durable replay. Records only insert or raise +// generations; this counter is what lets a scanner cycle tell the two apart +// across its bucket-listing window (`covers_listed_pending`). +static DIRTY_USAGE_REMOVAL_GENERATION: AtomicU64 = AtomicU64::new(0); pub(super) static DIRTY_USAGE_BUCKETS: LazyLock> = LazyLock::new(|| StdMutex::new(HashMap::new())); // Lock order when dirty usage state is updated is `DIRTY_USAGE_BUCKETS`, // `DIRTY_USAGE_BUCKET_SCOPES`, then `DIRTY_USAGE_PRODUCER_IDENTITIES`. All @@ -281,6 +286,7 @@ pub fn replay_durable_dirty_usage_producer_record( ); } DIRTY_USAGE_PRODUCER_COVERAGE.fetch_or(coverage, Ordering::AcqRel); + advance_generation(&DIRTY_USAGE_REMOVAL_GENERATION); dirty_buckets.len() }; global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending)); @@ -415,6 +421,7 @@ pub fn acknowledge_scoped_dirty_usage( if cleared > 0 { producer_identities.retain(|bucket, _| dirty.contains_key(bucket)); advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); + advance_generation(&DIRTY_USAGE_REMOVAL_GENERATION); } (cleared, dirty.len(), cleared_buckets) }; @@ -1196,6 +1203,7 @@ pub fn acknowledge_dirty_usage_generation( let cleared_buckets = before.saturating_sub(dirty_buckets.len()); if cleared_buckets > 0 { advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); + advance_generation(&DIRTY_USAGE_REMOVAL_GENERATION); } (cleared_buckets, dirty_buckets.len(), cleared) }; @@ -1226,6 +1234,9 @@ pub fn clear_dirty_usage_bucket(bucket: &str) { producer_identities.remove(bucket); DIRTY_USAGE_PRODUCER_COVERAGE.store(0, Ordering::Release); advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); + if cleared.is_some() { + advance_generation(&DIRTY_USAGE_REMOVAL_GENERATION); + } (dirty_buckets.len(), cleared) }; if let Some(cleared) = cleared { @@ -1234,8 +1245,59 @@ pub fn clear_dirty_usage_bucket(bucket: &str) { global_metrics().record_scanner_dirty_usage_clear(usize_to_u64_saturated(pending_buckets)); } +/// Dirty-usage state captured immediately before a cycle lists its buckets. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct DirtyUsageListingCutoff { + pub(super) generation: u64, + /// `None` when the caller did not capture removals; the snapshot then + /// admits scoping only under the strict `covers_all_pending` rule. + removal_generation: Option, +} + +pub(super) fn dirty_usage_listing_cutoff() -> DirtyUsageListingCutoff { + // Removals advance both counters under this lock, so reading them under + // it pairs the generation with every removal that precedes it. + let _dirty_buckets = dirty_usage_buckets(); + DirtyUsageListingCutoff { + generation: DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire), + removal_generation: Some(DIRTY_USAGE_REMOVAL_GENERATION.load(Ordering::Acquire)), + } +} + +#[cfg(test)] pub(super) fn snapshot_dirty_usage_buckets(buckets: &[BucketInfo], absent_generation_cutoff: u64) -> DirtyUsageSnapshot { - let (snapshot, scopes, generation, covers_all_pending) = { + snapshot_dirty_usage_buckets_after_listing( + buckets, + DirtyUsageListingCutoff { + generation: absent_generation_cutoff, + removal_generation: None, + }, + ) +} + +/// Snapshot the dirty map after the bucket listing taken at `cutoff`. +/// +/// The snapshot holds every entry recorded at or before the cutoff and every +/// later entry for a listed bucket. The only entries it leaves out are +/// buckets this cycle did not list that became dirty after the cutoff: the +/// walk cannot visit them, certification never clears them, and +/// `pending_dirty_usage_buckets` reports them, so the next cycle selects them. +/// +/// `covers_all_pending` keeps its strict meaning (nothing moved during the +/// listing) because segment-reuse evidence and `DirtyUsageSnapshotStatus` +/// bind to an unchanged generation window. `covers_listed_pending` is the +/// weaker property a bucket-scoped walk needs: no entry was removed or +/// replaced during the listing, so every bucket changed since the baseline +/// that this cycle can walk is still in the snapshot. A removal in the window +/// may be a certification against a publication newer than the baseline the +/// cycle loaded, and scoping would then reuse a stale baseline value for that +/// bucket, so any removal keeps the strict rule. +pub(super) fn snapshot_dirty_usage_buckets_after_listing( + buckets: &[BucketInfo], + cutoff: DirtyUsageListingCutoff, +) -> DirtyUsageSnapshot { + let absent_generation_cutoff = cutoff.generation; + let (snapshot, scopes, generation, covers_all_pending, covers_listed_pending) = { let dirty_buckets = dirty_usage_buckets(); let dirty_scopes = dirty_usage_bucket_scopes(); let listed_buckets = dirty_buckets @@ -1266,7 +1328,24 @@ pub(super) fn snapshot_dirty_usage_buckets(buckets: &[BucketInfo], absent_genera .collect::(); let generation = DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire); let covers_all_pending = generation == absent_generation_cutoff && snapshot.len() == dirty_buckets.len(); - (snapshot, scopes, generation, covers_all_pending) + let removal_generation = DIRTY_USAGE_REMOVAL_GENERATION.load(Ordering::Acquire); + let no_removal_since_cutoff = cutoff + .removal_generation + .is_some_and(|before| before == removal_generation && removal_generation != u64::MAX); + // The filter above already guarantees the second clause; it is + // restated so a change to the filter cannot silently drop a listed + // bucket from a scoped walk. + let covers_listed_pending = covers_all_pending + || (generation != u64::MAX + && no_removal_since_cutoff + && dirty_buckets.iter().all(|(bucket, dirty_generation)| { + snapshot.contains_key(bucket) + || (*dirty_generation > absent_generation_cutoff + && listed_buckets + .as_ref() + .is_some_and(|listed_buckets| !listed_buckets.contains(bucket.as_str()))) + })); + (snapshot, scopes, generation, covers_all_pending, covers_listed_pending) }; global_metrics().record_scanner_dirty_usage_cycle_snapshot(usize_to_u64_saturated(snapshot.len())); DirtyUsageSnapshot { @@ -1274,6 +1353,7 @@ pub(super) fn snapshot_dirty_usage_buckets(buckets: &[BucketInfo], absent_genera scopes: Arc::new(scopes), generation, covers_all_pending, + covers_listed_pending, } } @@ -1306,6 +1386,7 @@ pub(super) fn clear_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) { if cleared_buckets > 0 { producer_identities.retain(|bucket, _| dirty_buckets.contains_key(bucket)); advance_generation(&DIRTY_USAGE_BUCKET_GENERATION); + advance_generation(&DIRTY_USAGE_REMOVAL_GENERATION); } (cleared_buckets, dirty_buckets.len(), cleared) }; @@ -1465,6 +1546,7 @@ pub(super) fn dirty_usage_bucket_count() -> usize { #[cfg(test)] pub(crate) fn clear_dirty_usage_buckets_for_tests() { dirty_usage_buckets().clear(); + advance_generation(&DIRTY_USAGE_REMOVAL_GENERATION); dirty_usage_bucket_scopes().clear(); dirty_usage_producer_identities().clear(); DIRTY_USAGE_PRODUCER_COVERAGE.store(0, Ordering::Release); diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index db55aa91c..14876ee9d 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -135,7 +135,7 @@ where return default_result(ScannerBucketScanScope::default()); } if !resolution.requested_scope.is_default() - || !resolution.dirty_usage_snapshot.covers_all_pending + || !resolution.dirty_usage_snapshot.covers_listed_pending || resolution.dirty_usage_snapshot.generation == u64::MAX || resolution.dirty_usage_snapshot.buckets.len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES { @@ -264,6 +264,32 @@ where )) } +/// Runs once between the bucket listing and the dirty-usage snapshot, the +/// window in which a write lands after the listing cutoff. +#[cfg(test)] +type BucketListingWindowHook = Box; + +#[cfg(test)] +static BUCKET_LISTING_WINDOW_HOOK: StdMutex> = StdMutex::new(None); + +#[cfg(test)] +pub(super) fn set_bucket_listing_window_hook_for_tests(hook: impl FnOnce() + Send + 'static) { + *BUCKET_LISTING_WINDOW_HOOK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(Box::new(hook)); +} + +#[cfg(test)] +fn run_bucket_listing_window_hook_for_tests() { + let hook = BUCKET_LISTING_WINDOW_HOOK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); + if let Some(hook) = hook { + hook(); + } +} + #[cfg(test)] pub(super) async fn resolve_scanner_bucket_scan_scope_for_tests( store: &S, @@ -366,8 +392,10 @@ where )); } }; - let dirty_generation_before_bucket_list = dirty_usage_generation(); + let dirty_usage_cutoff = dirty_usage_listing_cutoff(); let bucket_listing = store.list_bucket_for_scanner(&BucketOptions::default()).await?; + #[cfg(test)] + run_bucket_listing_window_hook_for_tests(); let mut bucket_plan_complete = bucket_listing.topology_complete; let all_buckets = Arc::new(bucket_listing.buckets); let set_disks = store.all_set_disks(); @@ -400,24 +428,27 @@ where let checkpoint_coverage_digest = scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_coverage_digest(&activity_before)); let execution_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan); - let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list)); + let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets_after_listing(&all_buckets, dirty_usage_cutoff)); + let baseline_proof = ScannerCacheBaselineProof { + authoritative_data: persisted_usage_baseline.as_ref(), + observed_candidate_data: observed_usage_candidate.as_ref(), + expected_sources: &expected_sources, + leader_epoch, + want_cycle, + scan_plan_digest: structural_scan_plan_digest, + }; + let full_scan_required = requires_full_scan || scan_mode == HealScanMode::Deep; + let scope_eligible = !full_scan_required && scan_scope.is_default(); let scope_resolution = resolve_scanner_bucket_scan_scope( store, distributed, ScannerBucketScopeResolution { requested_scope: scan_scope, - baseline_proof: ScannerCacheBaselineProof { - authoritative_data: persisted_usage_baseline.as_ref(), - observed_candidate_data: observed_usage_candidate.as_ref(), - expected_sources: &expected_sources, - leader_epoch, - want_cycle, - scan_plan_digest: structural_scan_plan_digest, - }, + baseline_proof, activity_before: &activity_before, dirty_usage_snapshot: &dirty_usage_snapshot, all_buckets: &all_buckets, - requires_full_scan: requires_full_scan || scan_mode == HealScanMode::Deep, + requires_full_scan: full_scan_required, #[cfg(test)] test_peer_snapshots: None, #[cfg(test)] @@ -428,6 +459,29 @@ where let remote_dirty_usage_acknowledgements = scope_resolution.remote_dirty_usage_acknowledgements; let distributed_segment_invalidation_evidence = scope_resolution.distributed_segment_invalidation_evidence; let scan_scope = scope_resolution.scope; + let scope_fallback = scanner_scope_fallback_reason( + scope_eligible, + &scan_scope, + &dirty_usage_snapshot, + &all_buckets, + baseline_proof, + distributed, + ); + if let Some(reason) = scope_fallback { + global_metrics().record_scanner_scope_fallback(reason.as_str()); + tracing::info!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_SET_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "scoped_scan_fallback", + reason = reason.as_str(), + cycle = want_cycle, + dirty_buckets = dirty_usage_snapshot.buckets.len(), + bucket_count = all_buckets.len(), + "Scanner cycle walks every bucket because its dirty-bucket scope was not admitted" + ); + } let segment_invalidation_proof = dirty_usage_producer_evidence(&dirty_usage_snapshot).segment_invalidation_proof(); #[cfg(test)] if let Some(observer) = resolved_scope_observer { @@ -506,6 +560,7 @@ where .with_remote_publication_lease_targets(remote_publication_lease_targets) .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) .with_segment_reuse_activation_preflight(segment_reuse_activation_preflight) + .with_scope_fallback(scope_fallback) .with_publication_expectation(publication_expectation)); } @@ -881,5 +936,6 @@ where .with_failed_dirty_usage(!failed_buckets.is_empty()) .with_pending_maintenance_work(pending_maintenance_work) .with_required_cycle_floor(required_cycle_floor) + .with_scope_fallback(scope_fallback) .with_publication_expectation(publication_expectation)) } diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 29a5652d1..a1fd568d4 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -45,6 +45,7 @@ use time::OffsetDateTime; use uuid::Uuid; mod checkpoint_coverage_identity; +mod listing_window_scope; mod partial_bucket_carry; mod per_bucket_certification; mod scoped_entry_fallback; @@ -185,6 +186,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_reports_cycle_inputs_wit scopes: Arc::new(DirtyUsageBucketScopes::default()), generation: 7, covers_all_pending: true, + covers_listed_pending: true, }; let distributed_evidence = DistributedSegmentInvalidationEvidence { invalidation_domain: crate::segment_invalidation::SegmentInvalidationDomain::DistributedEc, @@ -218,6 +220,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_blocks_unbounded_inputs( scopes: Arc::new(DirtyUsageBucketScopes::default()), generation: u64::MAX, covers_all_pending: false, + covers_listed_pending: false, }; let preflight = scanner_segment_reuse_activation_preflight_for_cycle( @@ -251,6 +254,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke scopes: Arc::new(DirtyUsageBucketScopes::default()), generation: 7, covers_all_pending: true, + covers_listed_pending: true, }; let preflight = scanner_segment_reuse_activation_preflight_for_cycle( @@ -2835,6 +2839,7 @@ fn remote_dirty_usage_invalidates_local_prefix_hints_until_distributed_proof_exi scopes: Arc::new(dirty_scopes.clone()), generation: 7, covers_all_pending: true, + covers_listed_pending: true, }; let locally_scoped = scoped_scan_scope_from_dirty_buckets( ScannerBucketScanScope::default(), @@ -3153,6 +3158,7 @@ fn remote_dirty_usage_scope_resolution_falls_back_when_ack_batch_exceeds_thresho scopes: Arc::new(HashMap::new()), generation: 7, covers_all_pending: true, + covers_listed_pending: true, }, remote_dirty_usage, &all_buckets, @@ -3279,6 +3285,7 @@ async fn distributed_scoped_scan_falls_back_when_remote_scoped_ack_capability_is scopes: Arc::new(HashMap::new()), generation: 7, covers_all_pending: true, + covers_listed_pending: true, }; let activity_before = BTreeMap::from([( "node-a:9000".to_string(), diff --git a/crates/scanner/src/scanner_io/tests/listing_window_scope.rs b/crates/scanner/src/scanner_io/tests/listing_window_scope.rs new file mode 100644 index 000000000..4c4b40bbd --- /dev/null +++ b/crates/scanner/src/scanner_io/tests/listing_window_scope.rs @@ -0,0 +1,206 @@ +// Copyright 2026 RustFS Team +// SPDX-License-Identifier: Apache-2.0 + +//! Scope admission across the bucket-listing window on the single-disk +//! fixture, without lifecycle or replication rules. A write that lands in a +//! listed bucket between the listing cutoff and the dirty-usage snapshot is +//! part of the start snapshot and must not force a full scan; a dirty entry +//! removed in the same window still does. + +use super::*; + +const BUCKETS: [&str; 3] = ["cold-bucket", "hot-bucket", "idle-bucket"]; + +async fn put_fixture_object(store: &Arc, bucket: &str, object: &str) { + let mut reader = ScannerPutObjReader::from_vec(vec![7; 16]); + store + .put_object(bucket, object, &mut reader, &ScannerObjectOptions::default()) + .await + .expect("fixture object should persist"); +} + +async fn run_cycle( + store: &Arc, + cycle: u64, + baseline: Option, + requires_full_scan: bool, +) -> (ScannerCycleResult, DataUsageInfo, ScannerBucketScanScope) { + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + let (updates, mut receiver) = mpsc::channel(1); + let (observer, resolved_scope) = tokio::sync::oneshot::channel(); + let result = tokio::time::timeout( + Duration::from_secs(60), + nsscanner_with_storage_status_scoped( + store.as_ref(), + ScannerCycleRequest { + ctx, + budget, + updates, + want_cycle: cycle, + leader_epoch: 11, + scan_mode: HealScanMode::Normal, + scan_scope: ScannerBucketScanScope::default(), + persisted_usage_baseline: baseline, + observed_usage_candidate: None, + requires_full_scan, + service_cohort: None, + resolved_scope_observer: Some(observer), + post_walk_gate: None, + }, + ), + ) + .await + .expect("cycle should finish within the test deadline") + .expect("cycle should succeed"); + let snapshot = receiver.recv().await.expect("cycle should publish a snapshot"); + let scope = resolved_scope.await.expect("the cycle should report its resolved scope"); + (result, snapshot, scope) +} + +fn baseline_bytes(snapshot: &DataUsageInfo) -> Bytes { + Bytes::from(serde_json::to_vec(snapshot).expect("snapshot should encode")) +} + +/// Creates the fixture buckets and certifies them with one quiet full walk, +/// returning the converged snapshot as the next cycle's baseline. +async fn converged_baseline(store: &Arc) -> Bytes { + clear_dirty_usage_buckets_for_tests(); + for bucket in BUCKETS { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + put_fixture_object(store, bucket, "initial").await; + record_dirty_usage_bucket(bucket); + } + wait_for_namespace_commit_tails(store.as_ref()).await; + let (result, first, _) = run_cycle(store, 1, None, true).await; + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(first.usage_snapshot_converged, Some(true)); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + baseline_bytes(&first) +} + +#[tokio::test] +#[serial] +async fn write_to_a_listed_bucket_during_listing_keeps_the_cycle_scoped() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + let baseline = converged_baseline(&store).await; + + put_fixture_object(&store, "cold-bucket", "second").await; + record_dirty_usage_bucket("cold-bucket"); + put_fixture_object(&store, "hot-bucket", "second").await; + wait_for_namespace_commit_tails(store.as_ref()).await; + // The hot bucket's dirty mark lands after the listing cutoff but before + // the snapshot: under sustained writes some bucket is always in this + // window, so the listing must not be required to be globally quiet. + super::io_cycle::set_bucket_listing_window_hook_for_tests(|| record_dirty_usage_bucket("hot-bucket")); + + let (result, second, scope) = run_cycle(&store, 2, Some(baseline), false).await; + + assert_eq!( + scope.selected_buckets_for_tests(), + Some(&HashSet::from(["cold-bucket".to_string(), "hot-bucket".to_string()])), + "a listed bucket written during listing joins the start snapshot instead of forcing a full scan" + ); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(result.scope_fallback(), None); + assert_eq!(result.pending_dirty_usage_buckets(), 0); + assert_eq!(second.usage_snapshot_converged, Some(true)); + assert_eq!(second.buckets_usage["cold-bucket"].objects_count, 2); + assert_eq!(second.buckets_usage["hot-bucket"].objects_count, 2); + assert_eq!(second.buckets_usage["idle-bucket"].objects_count, 1); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn bucket_created_during_listing_stays_pending_without_forcing_a_full_scan() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + let baseline = converged_baseline(&store).await; + + put_fixture_object(&store, "cold-bucket", "second").await; + record_dirty_usage_bucket("cold-bucket"); + wait_for_namespace_commit_tails(store.as_ref()).await; + // A bucket the listing did not return cannot be certified by this cycle; + // it must stay grey and be reported pending, not silently cleared. + super::io_cycle::set_bucket_listing_window_hook_for_tests(|| record_dirty_usage_bucket("created-after-listing")); + + let (result, second, scope) = run_cycle(&store, 2, Some(baseline), false).await; + + assert_eq!(scope.selected_buckets_for_tests(), Some(&HashSet::from(["cold-bucket".to_string()]))); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(result.scope_fallback(), None); + assert_eq!(result.pending_dirty_usage_buckets(), 1); + assert_eq!(second.usage_snapshot_pending_buckets, vec!["created-after-listing".to_string()]); + assert!(result.clear_verified_usage().is_empty()); + assert_eq!( + dirty_usage_buckets_for_tests().keys().cloned().collect::>(), + vec!["created-after-listing".to_string()], + "the unlisted bucket stays dirty for the next cycle" + ); + clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn dirty_entry_removed_during_listing_still_forces_a_full_scan() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + let baseline = converged_baseline(&store).await; + + put_fixture_object(&store, "cold-bucket", "second").await; + record_dirty_usage_bucket("cold-bucket"); + put_fixture_object(&store, "hot-bucket", "second").await; + record_dirty_usage_bucket("hot-bucket"); + wait_for_namespace_commit_tails(store.as_ref()).await; + // A removal in the window may be a certification against a publication + // newer than the baseline this cycle loaded; scoping would then reuse a + // stale baseline value for that bucket, so the cycle stays a full scan. + // The unlisted bucket dirtied in the same window stays pending, which is + // what the leader loop paces instead of waking immediately. + super::io_cycle::set_bucket_listing_window_hook_for_tests(|| { + clear_dirty_usage_bucket("hot-bucket"); + record_dirty_usage_bucket("created-after-listing"); + }); + let fallbacks_before = global_metrics().report().await.usage_freshness.scope_fallback_total; + + let (result, second, scope) = run_cycle(&store, 2, Some(baseline), false).await; + + assert!(scope.is_default(), "a removal during listing must not admit a scoped scan"); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(result.scope_fallback(), Some(ScannerScopeFallbackReason::DirtySnapshotIncomplete)); + assert_eq!(result.pending_dirty_usage_buckets(), 1); + assert_eq!( + result.completed_scope_fallback(), + Some(ScannerScopeFallbackReason::DirtySnapshotIncomplete) + ); + assert_eq!(second.buckets_usage["hot-bucket"].objects_count, 2); + assert_eq!(second.buckets_usage["cold-bucket"].objects_count, 2); + let freshness = global_metrics().report().await.usage_freshness; + assert_eq!(freshness.scope_fallback_total, fallbacks_before + 1); + assert_eq!(freshness.last_scope_fallback_reason, "dirty_snapshot_incomplete"); + clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn required_full_scan_is_not_reported_as_a_scope_fallback() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + let baseline = converged_baseline(&store).await; + put_fixture_object(&store, "cold-bucket", "second").await; + record_dirty_usage_bucket("cold-bucket"); + wait_for_namespace_commit_tails(store.as_ref()).await; + super::io_cycle::set_bucket_listing_window_hook_for_tests(|| record_dirty_usage_bucket("created-after-listing")); + + let (result, _, scope) = run_cycle(&store, 2, Some(baseline), true).await; + + assert!(scope.is_default()); + assert_eq!(result.pending_dirty_usage_buckets(), 1); + assert_eq!(result.scope_fallback(), None, "scheduled maintenance walks are not degradations"); + assert_eq!(result.completed_scope_fallback(), None); + clear_dirty_usage_buckets_for_tests(); +} diff --git a/docs/architecture/scanner-per-bucket-convergence.md b/docs/architecture/scanner-per-bucket-convergence.md index 99f57f829..b45691c6f 100644 --- a/docs/architecture/scanner-per-bucket-convergence.md +++ b/docs/architecture/scanner-per-bucket-convergence.md @@ -137,6 +137,26 @@ Mutations after that point are the next cycle's work. This is what makes termination possible under continuous writes: the termination condition is "no grey bucket in the start snapshot", not "no mutation anywhere". +The start snapshot is read after the bucket listing, not before it. The +listing window is the gap between the listing cutoff (the dirty-usage +generation read before `list_bucket_for_scanner`) and that read; under +sustained writes some bucket is almost always written inside it, so the +scope cannot require the whole dirty map to be quiet across the listing. +`snapshot_dirty_usage_buckets_after_listing` keeps every entry recorded at or +before the cutoff and every later entry for a listed bucket. The only entries +it leaves out are buckets the listing did not return that became dirty after +the cutoff: the walk cannot visit them, certification compares generations +against the snapshot and so never clears them, and +`pending_dirty_usage_buckets` reports them, so the next cycle lists and +selects them. A scoped walk therefore needs only `covers_listed_pending`: no +dirty entry was removed or overwritten by replay during the listing. A +removal in the window may be a certification against a publication newer +than the baseline the cycle loaded, and scoping would then reuse a stale +baseline value for that bucket, so a removal still makes the cycle a full +walk. The strict `covers_all_pending` (nothing moved at all) keeps gating +segment-reuse evidence, which binds to an unchanged generation window; a +cycle admitted only by the weaker rule walks its dirty buckets whole. + Usage is a sum, not a reachability fixed point. A certified bucket's value plus an uncertified bucket's last certified value is a valid snapshot, so the collector does not need a globally consistent heap to publish. This is why a @@ -359,7 +379,7 @@ not tuning advice. |---|---| | Grey buckets are rescanned next cycle | A scoped cycle walks at most a configured number of buckets and objects. Grey buckets beyond that are ordered by waiting age, oldest first, so every bucket is walked within a bounded number of cycles. New names never reorder ahead of an older waiter. | | `Status` hits are not namespace activity | A `Status` hit costs one atomic OR on the writeback path and no scanner work. It never triggers a dirty-usage wake. A replication resync therefore adds nothing to scanner load beyond the full walk the cycle already performs. | -| A `Complete` cycle with pending buckets is a baseline | When the pending set is the whole namespace (a mass write, a disk loss), the next cycle is declared a full sweep explicitly and uses the existing superseded retry backoff schedule. A full sweep is never run under the scoped cycle's wake cadence. | +| A `Complete` cycle with pending buckets is a baseline | A cycle eligible for a dirty-bucket scope that walks the whole namespace instead (more dirty buckets than one scoped cycle may select, no usable baseline, a dirty entry removed during the listing, unverifiable peer dirty usage) is paced while dirty buckets remain after it: the next cycle waits on the capped retry schedule (5 s doubling) clamped to the configured cycle interval, with the dirty-usage wake disabled and cluster activity deferred. A full sweep is therefore never run under the scoped cycle's wake cadence, and a namespace that stays unscopable is still walked at the ordinary cycle cadence rather than starved. The paced timer wake still requests a scope, and any non-fallback cycle resets the schedule. | | Bounded remark | At most one remark pass per cycle, inside the remaining budget. | | Old-generation sweep every `M` cycles | The sweep is checkpointed and resumes; it does not restart on a `Namespace` hit. A hit only demotes the bucket to young for the following cycle. | | Pacer raises walkers and lowers sleep | CPU and I/O ceilings are absolute. Foreground admission backoff cannot be undercut. Walker count never exceeds the ceiling-derived maximum even when the target is missed. | @@ -482,15 +502,27 @@ condition before merging. dirty map holds the shaded pending buckets, and the published pending list also names buckets whose walk stopped early), the attributed guardrail veto (`guardrail_veto_total`, `last_guardrail_veto_unix_secs`, - `last_guardrail_veto_reason` as `@`), plus the pacer target, + `last_guardrail_veto_reason` as `@`), the scope fallback + (`scope_fallback_total`, `last_scope_fallback_unix_secs`, + `last_scope_fallback_reason`, one of `dirty_snapshot_incomplete`, + `dirty_generation_exhausted`, `dirty_bucket_overflow`, + `no_listed_dirty_buckets`, `baseline_unavailable`, + `peer_scope_unavailable`), plus the pacer target, achieved rate, and whether a ceiling is binding. The console marks usage stale only for that count or for a guardrail veto. - `last_cycle_result` reports `success`, its existing label for a completed cycle, for a cycle with pending buckets; `superseded` is reserved for guardrail vetoes and the phase-2 peer fallback and should be rare. -- The superseded retry schedule applies only to guardrail vetoes and to - explicit full sweeps. A `Complete` cycle with pending buckets uses the - ordinary dirty-usage wake and the next cycle is scoped to the pending set. +- The superseded retry schedule applies only to guardrail vetoes. A + `Complete` scoped cycle with pending buckets uses the ordinary dirty-usage + wake and the next cycle is scoped to the pending set. While dirty buckets + remain after a full-namespace scope fallback, the next cycle uses the + capped retry schedule clamped to the cycle interval (see + [Boundedness constraints](#boundedness-constraints)), and the fallback + logs `scoped_scan_fallback` with its reason. Cycles that must walk + everything regardless (lifecycle or replication rules, deep scans, the + first cycle after leadership) are not scope fallbacks and are not + reported as such. - `rustfs_cluster_usage_objects_snapshot_converged` keeps its meaning. ## Delivery