diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 01199f8a1..8e2b62a92 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -2740,6 +2740,7 @@ where Err(e) => { match classify_scanner_leader_lock_failure(&e) { ScannerLeaderLockFailure::Contended => { + clear_scanner_transient_cycle_recovery_status(); record_scanner_cycle_schedule_role("follower"); record_scanner_leader_lock_state("contended"); global_metrics().record_scanner_leader_liveness("contended", false, "").await; diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 720302608..f2801033b 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -353,6 +353,17 @@ fn publish_scanner_cleanup_failure(reason: String, expected: u64) { let _ = publish_scanner_cleanup_status(status, Some(expected)); } +pub(super) fn clear_scanner_transient_cycle_recovery_status() { + // Snapshot the version before reading status so any concurrent publish rejects the CAS below. + let version = scanner_cleanup_status_version(); + let status = scanner_cycle_recovery_status(); + // A follower can discard its failed leadership observation. Durable recovery + // markers are revalidated before any later leadership claim. + if status.classification.is_none() && matches!(status.state.as_str(), "transient" | "paused") { + let _ = publish_scanner_cleanup_status(recovery_status("healthy", None, false), Some(version)); + } +} + pub(super) fn record_scanner_usage_floor_failure(reason: String) { let previous = scanner_cycle_recovery_status(); let same_failure = previous.classification.as_deref() == Some(USAGE_FLOOR_LOAD_FAILED); diff --git a/crates/scanner/src/scanner/tests/recovery_control.rs b/crates/scanner/src/scanner/tests/recovery_control.rs index 61ac96bc5..b721435d1 100644 --- a/crates/scanner/src/scanner/tests/recovery_control.rs +++ b/crates/scanner/src/scanner/tests/recovery_control.rs @@ -81,6 +81,159 @@ async fn persisted_state(store: &Arc) -> Vec<(Option>, DataUsag state } +async fn seed_transient_cycle_read_failure() -> ScannerCycleRecoveryStatus { + let store = Arc::new(MemoryConfigStore::default()); + let error = EcstoreError::InsufficientReadQuorum(RUSTFS_META_BUCKET.to_string(), DATA_USAGE_BLOOM_NAME_PATH.clone()); + let expected_error = format!("Scanner error: failed to inspect scanner cycle state: {error}"); + store + .read_errors + .lock() + .await + .insert(memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()), error); + let ScannerCycleStateStartup::Transient(error) = load_scanner_cycle_state_for_startup(store).await else { + panic!("a quorum outage must produce transient cycle recovery"); + }; + assert_eq!(error.to_string(), expected_error); + let status = scanner_cycle_recovery_status(); + assert_eq!(status.state, "transient"); + assert!(status.retryable); + assert_eq!(status.classification, None); + status +} + +#[tokio::test] +#[serial] +async fn issue_8297_contended_follower_clears_only_transient_cycle_recovery() { + temp_env::async_with_vars([("RUSTFS_LOCK_ACQUIRE_TIMEOUT", Some("1"))], async { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "cleanup-pending").await; + let before = persisted_state(&store).await; + let lock = store + .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock") + .await + .expect("leader lock"); + let guard = lock + .get_write_lock_quiet(Duration::from_secs(1)) + .await + .expect("hold live leader"); + let transient = seed_transient_cycle_read_failure().await; + assert!(!record_scanner_cycle_recovery_retry(MAX_SCANNER_CYCLE_RECOVERY_RETRIES)); + let paused = scanner_cycle_recovery_status(); + assert_eq!(paused.state, "paused"); + let mut cases = vec![(transient, true), (paused, true)]; + for (state, classification, retryable) in [ + ("recovery-required", None, false), + ("blocked", None, false), + ("blocked", Some("corrupt"), false), + ("cleanup-pending", None, true), + ("cleanup-pending", Some("corrupt"), true), + ("usage_floor_load_failed", Some("usage_floor_load_failed"), true), + ("usage_floor_recovery_pending", Some("legacy_empty_usage_floor"), true), + ("paused", Some("usage_floor_load_failed"), false), + ("cache_cycle_ahead", Some("cache_cycle_ahead"), true), + ] { + cases.push(( + ScannerCycleRecoveryStatus { + state: state.to_string(), + classification: classification.map(str::to_string), + retry_count: 5, + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + retryable, + reason: Some("preserve recovery state".to_string()), + ..Default::default() + }, + false, + )); + } + for (status, clear) in cases { + let expected = serde_json::to_value(&status).expect("expected recovery status"); + publish_scanner_cleanup_status(status, None).expect("seed local recovery status"); + tokio::time::timeout( + Duration::from_secs(30), + run_data_scanner_with_maintenance_state( + CancellationToken::new(), + store.clone(), + ScannerMaintenanceFeatures::default(), + None, + ), + ) + .await + .expect("contended scanner iteration must finish") + .expect("lock contention is a normal follower outcome"); + assert_eq!(scanner_cycle_schedule_status().execution_role, "follower"); + let actual = scanner_cycle_recovery_status(); + if clear { + assert_eq!(actual.state, "healthy", "stale follower recovery: {expected}"); + assert_eq!(actual.retry_count, 0); + assert!(!actual.retryable); + } else { + assert_eq!(serde_json::to_value(actual).expect("actual recovery status"), expected); + } + assert!(!guard.is_lock_lost(), "follower must not revoke the leader"); + assert_eq!(persisted_state(&store).await, before, "follower must not rewrite durable recovery state"); + } + }) + .await; +} + +#[tokio::test] +#[serial] +async fn issue_8297_supervisor_recovers_paused_follower_without_restart() { + temp_env::async_with_vars( + [ + ("RUSTFS_LOCK_ACQUIRE_TIMEOUT", Some("1")), + (ENV_SCANNER_START_DELAY_SECS, Some("0")), + (ENV_SCANNER_CYCLE, Some("1")), + ], + async { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "cleanup-pending").await; + let before = persisted_state(&store).await; + let lock = store + .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock") + .await + .expect("leader lock"); + let guard = lock + .get_write_lock_quiet(Duration::from_secs(1)) + .await + .expect("hold live leader"); + for attempt in 1..=MAX_SCANNER_CYCLE_RECOVERY_RETRIES { + seed_transient_cycle_read_failure().await; + assert_eq!(record_scanner_cycle_recovery_retry(attempt), attempt < MAX_SCANNER_CYCLE_RECOVERY_RETRIES); + } + assert_eq!(scanner_cycle_recovery_status().state, "paused"); + assert!(!crate::scanner_runtime_initialized(), "test must own the scanner runtime"); + let ctx = CancellationToken::new(); + let _cancel_on_drop = ctx.clone().drop_guard(); + init_data_scanner_with_storage(ctx.clone(), store.clone()).await; + assert!(crate::scanner_runtime_initialized()); + let recovered = tokio::time::timeout(Duration::from_secs(30), async { + while scanner_cycle_recovery_status().state != "healthy" { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + ctx.cancel(); + tokio::time::timeout(Duration::from_secs(15), async { + while crate::scanner_runtime_initialized() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("cancelled scanner supervisor must stop"); + recovered.expect("follower must recover without another five-minute pause or process restart"); + assert_eq!(scanner_cycle_schedule_status().execution_role, "follower"); + let status = scanner_cycle_recovery_status(); + assert_eq!(status.retry_count, 0); + assert!(!status.retryable); + assert!(!guard.is_lock_lost(), "supervisor must not revoke the leader"); + assert_eq!(persisted_state(&store).await, before, "supervisor must leave durable recovery untouched"); + }, + ) + .await; + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); +} + async fn run_disabled_startup(ctx: CancellationToken, store: Arc) { let initialized_before = crate::scanner_runtime_initialized(); let cleanup = init_scanner_with_recovery(ctx, store, false).await; diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index 372cce791..2029bc7d7 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -130,6 +130,7 @@ The request must be authenticated with an admin identity that has `ServerInfoAdm |---|---| | `runtime_config` | Effective runtime controls and their value sources. | | `cycle_schedule` | Current effective cycle interval and clean-idle backoff state. | +| `cycle_recovery` | This node's cycle-state recovery observation, retry budget, and reason. | | `metrics` | Scanner work, pressure, checkpoint, lifecycle, replication, heal, bitrot, and alert counters. | | `data_movement_pause` | Global-pause policy, current movement reason, operation epoch, start time, duration, and estimated movement work items. | | `pause_backlog` | Replicated durable pause ledger, post-pause catch-up phase, rate window, retry state, thresholds, and active alert reasons. | @@ -190,6 +191,12 @@ Shared compacted accumulators defer periodic emission until the owning subtree c A confirmed periodic save does not prove restart resumability. Adoption still validates bucket/set identity and, for non-empty caches, connected cache scope. Skipping completed subtrees additionally requires a validated coverage frontier; raw enumeration state alone does not prove completed subtree coverage. +### Cycle recovery on followers + +Cycle recovery status is local to each node. A transient cycle-state read failure can exhaust the short retry budget and enter `paused`; the enabled scanner then retries leadership every five minutes or when recovery is explicitly woken. If the next lock attempt is contended or times out waiting for `leader.lock`, the node records a follower iteration and clears only its unclassified `transient` or `paused` observation. Normal follower retry cadence resumes without restarting the process. + +This clears local retry bookkeeping, not persisted cycle state or recovery markers. Classified recovery states, cleanup-pending resets, and usage-floor failures remain unchanged. Any later leadership claim revalidates durable cycle state before scanning. A healthy follower recovery status does not prove that leader-dispatched scans are progressing; inspect the work counters separately. + ## Usage State Reset The supported break-glass route for rebuilding scanner usage state is `POST /v3/scanner/usage-state/reset` with body `{"mode":"full-rebuild"}`, authenticated as an admin identity holding `ConfigUpdateAdminAction` (route registered in `rustfs/src/admin/route_registration_test.rs`). Use it only after the scanner status shows a usage-floor load failure, a conflicting persisted usage floor, or an operator decision to discard the durable usage baseline and rebuild it from a full scanner pass.