mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 06:18:43 +00:00
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 <anzhengchao@gmail.com>
This commit is contained in:
唐小鸭
@@ -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@<local>".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@<local>");
|
||||
assert_eq!(scanner.usage_freshness.dirty_pending_buckets, 3);
|
||||
assert_eq!(scanner.usage_freshness.last_dirty_mark_unix_secs, 10);
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<String>,
|
||||
scanner_scope_fallback_total: AtomicU64,
|
||||
scanner_scope_fallback_last_unix_secs: AtomicU64,
|
||||
scanner_scope_fallback_last_reason: Mutex<String>,
|
||||
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<String>) {
|
||||
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;
|
||||
|
||||
@@ -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<Arc<StdMutex<crate::scanner_io::ScannerServiceCohort>>>,
|
||||
/// Receives the reason when a `Complete` cycle walked every bucket in
|
||||
/// place of an eligible dirty-bucket scope.
|
||||
scope_fallback: Option<Arc<ScannerScopeFallbackSlot>>,
|
||||
}
|
||||
|
||||
/// 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<Option<crate::scanner_io::ScannerScopeFallbackReason>>);
|
||||
|
||||
impl ScannerScopeFallbackSlot {
|
||||
fn set(&self, reason: Option<crate::scanner_io::ScannerScopeFallbackReason>) {
|
||||
*self.0.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = reason;
|
||||
}
|
||||
|
||||
fn take(&self) -> Option<crate::scanner_io::ScannerScopeFallbackReason> {
|
||||
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;
|
||||
|
||||
@@ -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<Duration> {
|
||||
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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -127,7 +127,11 @@ struct DirtyUsageSnapshot {
|
||||
buckets: Arc<DirtyUsageBuckets>,
|
||||
scopes: Arc<DirtyUsageBucketScopes>,
|
||||
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<ScannerScopeFallbackReason> {
|
||||
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::<HashSet<_>>();
|
||||
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<u64>,
|
||||
scope_fallback: Option<ScannerScopeFallbackReason>,
|
||||
publication_expectation: Option<ScannerPublicationExpectation>,
|
||||
}
|
||||
|
||||
@@ -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<ScannerScopeFallbackReason>) -> 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<ScannerScopeFallbackReason> {
|
||||
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<ScannerScopeFallbackReason> {
|
||||
(self.status == ScannerCycleStatus::Complete)
|
||||
.then_some(self.scope_fallback)
|
||||
.flatten()
|
||||
}
|
||||
|
||||
fn with_required_cycle_floor(mut self, required_cycle_floor: Option<u64>) -> Self {
|
||||
self.publication_expectation = None;
|
||||
self.required_cycle_floor = required_cycle_floor;
|
||||
|
||||
@@ -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<StdMutex<DirtyUsageBuckets>> = 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<u64>,
|
||||
}
|
||||
|
||||
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::<DirtyUsageBucketScopes>();
|
||||
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);
|
||||
|
||||
@@ -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<dyn FnOnce() + Send>;
|
||||
|
||||
#[cfg(test)]
|
||||
static BUCKET_LISTING_WINDOW_HOOK: StdMutex<Option<BucketListingWindowHook>> = 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<S>(
|
||||
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))
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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<ECStore>, 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<ECStore>,
|
||||
cycle: u64,
|
||||
baseline: Option<Bytes>,
|
||||
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<ECStore>) -> 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<_>>(),
|
||||
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();
|
||||
}
|
||||
@@ -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 `<field>@<host>`), plus the pacer target,
|
||||
`last_guardrail_veto_reason` as `<field>@<host>`), 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
|
||||
|
||||
Reference in New Issue
Block a user