fix(scanner): clear transient cycle recovery on follower lock contention (#8566)

This commit is contained in:
Ramakrishna Chilaka
2026-10-15 16:06:48 +05:30
committed by GitHub
parent 2c2c1133d4
commit 606d8a3a8a
4 changed files with 172 additions and 0 deletions
+1
View File
@@ -2740,6 +2740,7 @@ where
Err(e) => { Err(e) => {
match classify_scanner_leader_lock_failure(&e) { match classify_scanner_leader_lock_failure(&e) {
ScannerLeaderLockFailure::Contended => { ScannerLeaderLockFailure::Contended => {
clear_scanner_transient_cycle_recovery_status();
record_scanner_cycle_schedule_role("follower"); record_scanner_cycle_schedule_role("follower");
record_scanner_leader_lock_state("contended"); record_scanner_leader_lock_state("contended");
global_metrics().record_scanner_leader_liveness("contended", false, "").await; global_metrics().record_scanner_leader_liveness("contended", false, "").await;
+11
View File
@@ -353,6 +353,17 @@ fn publish_scanner_cleanup_failure(reason: String, expected: u64) {
let _ = publish_scanner_cleanup_status(status, Some(expected)); 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) { pub(super) fn record_scanner_usage_floor_failure(reason: String) {
let previous = scanner_cycle_recovery_status(); let previous = scanner_cycle_recovery_status();
let same_failure = previous.classification.as_deref() == Some(USAGE_FLOOR_LOAD_FAILED); let same_failure = previous.classification.as_deref() == Some(USAGE_FLOOR_LOAD_FAILED);
@@ -81,6 +81,159 @@ async fn persisted_state(store: &Arc<ECStore>) -> Vec<(Option<Vec<u8>>, DataUsag
state 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<ECStore>) { async fn run_disabled_startup(ctx: CancellationToken, store: Arc<ECStore>) {
let initialized_before = crate::scanner_runtime_initialized(); let initialized_before = crate::scanner_runtime_initialized();
let cleanup = init_scanner_with_recovery(ctx, store, false).await; let cleanup = init_scanner_with_recovery(ctx, store, false).await;
@@ -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. | | `runtime_config` | Effective runtime controls and their value sources. |
| `cycle_schedule` | Current effective cycle interval and clean-idle backoff state. | | `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. | | `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. | | `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. | | `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. 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 ## 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. 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.