diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 28cf638e9..7cc6d22e8 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -85,8 +85,8 @@ pub use rustfs_scanner_metrics::last_minute; pub use scanner::{ ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogStatus, ScannerPauseBacklogThresholds, ScannerUsageStateResetResult, - init_data_scanner, reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, scanner_cycle_recovery_status, - scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_topology_digest, + init_data_scanner, init_scanner_with_recovery, reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, + scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_topology_digest, }; pub use scanner_io::{ ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index b9de1c073..e3d461805 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -948,6 +948,33 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { init_data_scanner_with_storage(ctx, storeapi).await; } +/// Start normal scanning when enabled, or one resume-only cleanup attempt. +/// The disabled branch returns a finite task for the startup owner to join; +/// it never enables ordinary namespace scanning or accepts a new reset intent. +pub async fn init_scanner_with_recovery( + ctx: CancellationToken, + storeapi: Arc, + enabled: bool, +) -> Option> { + if enabled { + init_data_scanner(ctx, storeapi).await; + return None; + } + Some(tokio::spawn(async move { + if let Err(error) = resume_scanner_cycle_cleanup(ctx, storeapi).await { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "disabled_cleanup_deferred", + error = %error, + "Disabled scanner cleanup remains pending for an operator retry" + ); + } + })) +} + async fn init_data_scanner_with_storage(ctx: CancellationToken, storeapi: Arc) where S: ScannerStorage, diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 6913ed836..dfa9b8cdf 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -18,6 +18,7 @@ use crate::data_usage_define::{ DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_RECOVERY_PATH, usage_floor_primary_read_error_allows_backup, }; use crate::storage_api::owner::ObjectIO as _; +use std::sync::atomic::AtomicU64; use tokio::io::AsyncReadExt as _; const SCANNER_CYCLE_RECOVERY_SCHEMA_VERSION: u16 = 1; @@ -34,6 +35,86 @@ const CACHE_CYCLE_AHEAD: &str = "cache_cycle_ahead"; const SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD: &str = "full-rebuild"; +#[cfg(test)] +pub(super) mod cleanup_io_fault { + use super::*; + + #[derive(Clone, Copy, PartialEq, Eq)] + pub(in crate::scanner) enum Stage { + PrimaryRead, + PrimaryWrite, + UsageFence, + } + + struct Injection { + store: std::sync::Weak, + stage: Stage, + fired: AtomicBool, + owned: AtomicBool, + newer_completion: bool, + } + + static INJECTION: StdMutex>> = StdMutex::new(None); + pub(in crate::scanner) struct Guard(Arc); + + impl Guard { + pub(in crate::scanner) fn fired_while_owned(&self) -> bool { + self.0.fired.load(Ordering::Relaxed) && self.0.owned.load(Ordering::Relaxed) + } + } + + impl Drop for Guard { + fn drop(&mut self) { + let mut slot = INJECTION.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if slot.as_ref().is_some_and(|current| Arc::ptr_eq(current, &self.0)) { + *slot = None; + } + } + } + + pub(in crate::scanner) fn install(store: &Arc, stage: Stage, newer_completion: bool) -> Guard { + let injection = Arc::new(Injection { + store: Arc::downgrade(store), + stage, + fired: AtomicBool::new(false), + owned: AtomicBool::new(false), + newer_completion, + }); + let mut slot = INJECTION.lock().expect("cleanup injection slot"); + assert!(slot.is_none(), "only one cleanup I/O injection may be installed"); + *slot = Some(injection.clone()); + Guard(injection) + } + + pub(super) fn check(store: &Arc, stage: Stage, owned: bool) -> Result<(), ScannerError> { + let injection = { + let mut slot = INJECTION.lock().expect("cleanup injection slot"); + if slot + .as_ref() + .is_some_and(|injection| injection.stage == stage && injection.store.ptr_eq(&Arc::downgrade(store))) + { + slot.take() + } else { + None + } + }; + let Some(injection) = injection else { + return Ok(()); + }; + injection.fired.store(true, Ordering::Relaxed); + injection.owned.store(owned, Ordering::Relaxed); + if injection.newer_completion { + set_scanner_cycle_recovery_status(recovery_status("healthy", None, false)); + } + let reason = match stage { + Stage::PrimaryRead => "injected primary read failure", + Stage::PrimaryWrite => "injected primary write failure", + Stage::UsageFence => "injected usage fence failure", + }; + Err(ScannerError::Io(std::io::Error::other(reason))) + } +} + #[derive(Clone, Debug, Default, Serialize)] pub struct ScannerCycleRecoveryStatus { /// The immutable primary object whose revision is being guarded. @@ -81,6 +162,8 @@ static SCANNER_CYCLE_RECOVERY_STATUS: LazyLock ScannerCycleRecoveryStatus { SCANNER_CYCLE_RECOVERY_STATUS @@ -90,6 +173,24 @@ pub fn scanner_cycle_recovery_status() -> ScannerCycleRecoveryStatus { } fn set_scanner_cycle_recovery_status(status: ScannerCycleRecoveryStatus) { + let _ = publish_scanner_cleanup_status(status, None); +} + +pub(super) fn scanner_cleanup_status_version() -> u64 { + let _status = SCANNER_CYCLE_RECOVERY_STATUS + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + SCANNER_CYCLE_RECOVERY_STATUS_VERSION.load(Ordering::Relaxed) +} + +pub(super) fn publish_scanner_cleanup_status(status: ScannerCycleRecoveryStatus, expected: Option) -> Option { + let mut current = SCANNER_CYCLE_RECOVERY_STATUS + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let version = SCANNER_CYCLE_RECOVERY_STATUS_VERSION.load(Ordering::Relaxed); + if expected.is_some_and(|expected| expected == u64::MAX || expected != version) { + return None; + } let recovery_required = if matches!( status.state.as_str(), "blocked" @@ -106,9 +207,27 @@ fn set_scanner_cycle_recovery_status(status: ScannerCycleRecoveryStatus) { }; metrics::gauge!(METRIC_SCANNER_CYCLE_RECOVERY_REQUIRED).set(recovery_required); metrics::gauge!(METRIC_SCANNER_CYCLE_RECOVERY_RETRY_COUNT).set(status.retry_count as f64); - *SCANNER_CYCLE_RECOVERY_STATUS - .write() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = status; + *current = status; + let next = version.saturating_add(1); + SCANNER_CYCLE_RECOVERY_STATUS_VERSION.store(next, Ordering::Relaxed); + Some(next) +} + +fn publish_scanner_cleanup_failure(reason: String, expected: u64) { + let mut status = { + let current = SCANNER_CYCLE_RECOVERY_STATUS + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if SCANNER_CYCLE_RECOVERY_STATUS_VERSION.load(Ordering::Relaxed) != expected { + return; + } + current.clone() + }; + // Preserve the latest core progress, including a newly written primary's + // revision and epoch. The original marker may predate that durable write. + status.reason = Some(reason); + status.last_attempt_at_unix_secs = Some(unix_now_secs()); + let _ = publish_scanner_cleanup_status(status, Some(expected)); } pub(super) fn record_scanner_usage_floor_failure(reason: String) { @@ -932,10 +1051,81 @@ pub(crate) async fn load_scanner_cycle_state_for_startup( } } -/// Reset a blocked cycle state after an operator has explicitly requested a full -/// usage rebuild. The primary object is changed first with its observed ETag; -/// the recovery marker is removed only when its own ETag still matches. +/// Reset a blocked cycle state after an explicit full-rescan request. Durable +/// cleanup state fences primary rewrites; marker removal retains its ETag and +/// usage-epoch checks. pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc) -> Result<(), ScannerError> { + reset_scanner_cycle_recovery_for_intent(ctx, storeapi, None, None).await +} + +/// Resume only an operator reset whose cleanup phase is already durable. +/// Missing or merely blocked markers never authorize an automatic reset. +pub(super) async fn resume_scanner_cycle_cleanup(ctx: CancellationToken, storeapi: Arc) -> Result<(), ScannerError> { + let status_version = scanner_cleanup_status_version(); + let (marker, revision) = match read_scanner_cleanup_marker(storeapi.clone(), &ctx).await { + Ok(Some(marker)) => marker, + Ok(None) => return Ok(()), + Err(error) => { + let _ = + publish_scanner_cleanup_status(recovery_status("blocked", Some(&error.to_string()), false), Some(status_version)); + return Err(error); + } + }; + let mut observation = + publish_scanner_cleanup_status(recovery_status_from_marker(&marker, &marker.state), Some(status_version)); + if marker.state != "cleanup-pending" { + return Ok(()); + } + let result = reset_scanner_cycle_recovery_for_intent(ctx, storeapi, Some(revision), Some(&mut observation)).await; + if let Err(error) = &result + && let Some(observation) = observation + { + publish_scanner_cleanup_failure(error.to_string(), observation); + } + result +} + +pub(super) async fn read_scanner_cleanup_marker( + storeapi: Arc, + ctx: &CancellationToken, +) -> Result, ScannerError> { + let (data, revision) = tokio::select! { + biased; + _ = ctx.cancelled() => return Err(ScannerError::Other("scanner cleanup recovery was cancelled".to_string())), + result = tokio::time::timeout(data_usage_persist_timeout(), read_cycle_recovery_marker_bytes(storeapi)) => { + result.map_err(|_| ScannerError::Other("scanner cleanup marker inspection timed out".to_string()))? + .map_err(|err| ScannerError::Other(format!("failed to inspect pending scanner cleanup: {err}")))? + } + }; + let Some(data) = data else { + return Ok(None); + }; + let marker: ScannerCycleRecoveryMarker = serde_json::from_slice(&data) + .map_err(|err| ScannerError::Other(format!("pending scanner cleanup marker is invalid: {err}")))?; + validate_recovery_marker(&marker) + .map_err(|err| ScannerError::Other(format!("pending scanner cleanup marker is invalid: {err}")))?; + Ok(Some((marker, revision))) +} + +pub(super) async fn reset_scanner_cycle_recovery_for_intent( + ctx: CancellationToken, + storeapi: Arc, + expected_cleanup_revision: Option, + mut observation: Option<&mut Option>, +) -> Result<(), ScannerError> { + #[cfg(test)] + let resume_only = expected_cleanup_revision.is_some(); + // The outer Some distinguishes a tracked resume whose version may be + // invalidated from an explicit v3 reset with no observation owner. + let mut publish_status = |status| { + if let Some(version) = observation.as_deref_mut() { + if let Some(expected) = *version { + *version = publish_scanner_cleanup_status(status, Some(expected)); + } + } else { + set_scanner_cycle_recovery_status(status); + } + }; let lock = storeapi .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock") .await @@ -964,6 +1154,21 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< } Err(err) => return Err(ScannerError::Other(format!("failed to read cycle recovery marker: {err}"))), }; + if let Some(expected) = expected_cleanup_revision { + if marker_revision != expected || !owns_reset() { + return Err(ScannerError::Other( + "pending scanner cleanup changed before recovery acquired ownership".to_string(), + )); + } + let marker: ScannerCycleRecoveryMarker = marker_data + .as_deref() + .and_then(|data| serde_json::from_slice(data).ok()) + .filter(|marker| validate_recovery_marker(marker).is_ok() && marker.state == "cleanup-pending") + .ok_or_else(|| { + ScannerError::Other("scanner cleanup recovery requires an unchanged cleanup-pending marker".to_string()) + })?; + publish_status(recovery_status_from_marker(&marker, "cleanup-pending")); + } let Some(marker_data) = marker_data else { // A delete may commit before its reply is lost. Confirm both durable // fences before treating a retry without its marker as completed. @@ -981,7 +1186,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< "scanner cycle recovery marker is absent without a completed reset fence".to_string(), )); } - set_scanner_cycle_recovery_status(recovery_status("healthy", None, false)); + publish_status(recovery_status("healthy", None, false)); super::notify_scanner_cycle_recovery_wake(); return Ok(()); }; @@ -997,6 +1202,10 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< )); } + #[cfg(test)] + if resume_only { + cleanup_io_fault::check(&storeapi, cleanup_io_fault::Stage::PrimaryRead, owns_reset())?; + } let (mut primary_reader, primary_revision) = match storeapi .get_object_reader( RUSTFS_META_BUCKET, @@ -1062,7 +1271,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< let (cleanup_marker, cleanup_marker_revision) = mark_cycle_recovery_cleanup_pending(storeapi.clone(), marker.clone(), &marker_revision, reset_epoch, &owns_reset) .await?; - set_scanner_cycle_recovery_status(recovery_status_from_marker(&cleanup_marker, "cleanup-pending")); + publish_status(recovery_status_from_marker(&cleanup_marker, "cleanup-pending")); let usage_floor = persisted_usage_floor(storeapi.clone()).await?; let fence_epoch = primary_epoch .max(usage_floor.leader_epoch) @@ -1082,6 +1291,10 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< )); } verify_cycle_reset_intent(storeapi.clone(), &cleanup_marker_revision, &owns_reset).await?; + #[cfg(test)] + if resume_only { + cleanup_io_fault::check(&storeapi, cleanup_io_fault::Stage::PrimaryWrite, owns_reset())?; + } let preserved_info = save_reset_config( storeapi.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), @@ -1155,7 +1368,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< ScannerError::Other(format!("failed to clear stale cycle recovery marker: {err}")) } })?; - set_scanner_cycle_recovery_status(recovery_status("healthy", None, false)); + publish_status(recovery_status("healthy", None, false)); super::notify_scanner_cycle_recovery_wake(); return Ok(()); } else if !force_full_rescan && !marker_cleanup_pending { @@ -1228,11 +1441,23 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< )); } verify_cycle_reset_intent(storeapi.clone(), &marker_revision, &owns_reset).await?; - if let Err(err) = - fence_scanner_usage_epoch_with_expected_epoch(&ctx, storeapi.clone(), leader_epoch, Some(reset_epoch), false, &owns_reset) - .await - { - set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + let usage_fence = fence_scanner_usage_epoch_with_expected_epoch( + &ctx, + storeapi.clone(), + leader_epoch, + Some(reset_epoch), + false, + &owns_reset, + ); + #[cfg(test)] + let usage_fence = async { + if resume_only { + cleanup_io_fault::check(&storeapi, cleanup_io_fault::Stage::UsageFence, owns_reset())?; + } + usage_fence.await + }; + if let Err(err) = usage_fence.await { + publish_status(ScannerCycleRecoveryStatus { path: DATA_USAGE_BLOOM_NAME_PATH.clone(), quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), state: "cleanup-pending".to_string(), @@ -1260,7 +1485,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< } }; if current_revision != rebuilt_revision { - set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + publish_status(ScannerCycleRecoveryStatus { path: DATA_USAGE_BLOOM_NAME_PATH.clone(), quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), state: "cleanup-pending".to_string(), @@ -1281,7 +1506,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< } if guard.is_lock_lost() { - set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + publish_status(ScannerCycleRecoveryStatus { path: DATA_USAGE_BLOOM_NAME_PATH.clone(), quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), state: "cleanup-pending".to_string(), @@ -1318,7 +1543,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< .await { if scanner_publication_epoch_changed(&err) { - set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + publish_status(ScannerCycleRecoveryStatus { path: DATA_USAGE_BLOOM_NAME_PATH.clone(), quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), state: "cleanup-pending".to_string(), @@ -1336,7 +1561,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< "scanner recovery reset deferred by a movement epoch change".to_string(), )); } - set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + publish_status(ScannerCycleRecoveryStatus { path: DATA_USAGE_BLOOM_NAME_PATH.clone(), quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), state: "cleanup-pending".to_string(), @@ -1352,7 +1577,7 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< }); return Err(ScannerError::Other(format!("failed to clear cycle recovery marker: {err}"))); } - set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + publish_status(ScannerCycleRecoveryStatus { path: DATA_USAGE_BLOOM_NAME_PATH.clone(), quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), state: "healthy".to_string(), diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 37d45aa89..e89253d05 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -36,6 +36,8 @@ use tokio::time::{Duration, advance}; const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60; +mod recovery_control; + async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc) { setup_scanner_cycle_store_with_usage_baseline(true).await } diff --git a/crates/scanner/src/scanner/tests/recovery_control.rs b/crates/scanner/src/scanner/tests/recovery_control.rs new file mode 100644 index 000000000..ee76264c8 --- /dev/null +++ b/crates/scanner/src/scanner/tests/recovery_control.rs @@ -0,0 +1,475 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::super::cycle_state::cleanup_io_fault; +use super::*; +use crate::storage_api::owner::{EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats}; + +async fn seed_cleanup(store: &Arc, state: &str) -> ScannerCycleRecoveryMarker { + let cycle = CurrentCycle { + current: 3, + next: 42, + ..Default::default() + }; + save_config( + store.clone(), + DATA_USAGE_BLOOM_NAME_PATH.as_str(), + encode_scanner_cycle_state(&cycle, 7).expect("cycle encoding"), + ) + .await + .expect("persist cycle"); + let usage = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(41), + ..complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0) + }; + save_config( + store.clone(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + serde_json::to_vec(&usage).expect("usage encoding"), + ) + .await + .expect("persist usage floor"); + let marker = ScannerCycleRecoveryMarker { + schema_version: 1, + primary_revision: "previous-primary".to_string(), + generation: 41, + leader_epoch: 7, + classification: "corrupt".to_string(), + first_detected_at_unix_secs: 1, + last_attempt_at_unix_secs: 2, + retry_count: 1, + reason: "operator reset in progress".to_string(), + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(), + state: state.to_string(), + }; + save_config( + store.clone(), + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + serde_json::to_vec(&marker).expect("marker encoding"), + ) + .await + .expect("persist operator marker"); + marker +} + +async fn persisted_state(store: &Arc) -> Vec<(Option>, DataUsageCacheRevision)> { + let mut state = Vec::new(); + for path in [ + DATA_USAGE_BLOOM_NAME_PATH.as_str(), + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + ] { + state.push( + read_config_with_revision(store.clone(), path) + .await + .expect("read exact metadata revision"), + ); + } + state +} + +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; + if let Some(cleanup) = cleanup { + tokio::time::timeout(Duration::from_secs(15), cleanup) + .await + .expect("finite disabled cleanup attempt") + .expect("cleanup task should not panic"); + } + assert_eq!( + crate::scanner_runtime_initialized(), + initialized_before, + "disabled recovery must not start the normal scanner runtime" + ); +} + +async fn assert_reset_fences(store: &Arc) { + let data = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()) + .await + .expect("cycle remains durable"); + let (cycle, epoch) = decode_scanner_cycle_state(&data).expect("valid preserved cycle"); + assert_eq!((cycle.current, cycle.next, epoch), (3, 42, 8)); + let usage: DataUsageInfo = serde_json::from_slice( + &read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .expect("durable usage fence"), + ) + .expect("valid usage"); + assert_eq!(usage.scanner_epoch, Some(8)); + assert_eq!(usage.scanner_cycle, Some(41)); + assert!(matches!( + read_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await, + Err(EcstoreError::ConfigNotFound) + )); +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_reopens_persisted_intent_without_starting_scanner() { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "cleanup-pending").await; + let restarted = restart_scanner_cycle_store_from(&store).await; + run_disabled_startup(CancellationToken::new(), restarted.clone()).await; + assert_reset_fences(&restarted).await; + let completed = persisted_state(&restarted).await; + run_disabled_startup(CancellationToken::new(), restarted.clone()).await; + assert_eq!( + persisted_state(&restarted).await, + completed, + "a later startup without an intent must not reset again" + ); +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_does_not_authorize_blocked_unknown_or_corrupt_markers() { + let (_dir, store) = setup_scanner_cycle_store().await; + for kind in ["blocked", "unknown-phase", "future-version", "unknown-field", "corrupt"] { + let marker = seed_cleanup(&store, "blocked").await; + let mut value = serde_json::to_value(marker).expect("marker value"); + match kind { + "unknown-phase" => value["state"] = "future-phase".into(), + "future-version" => value["schema_version"] = 99.into(), + "unknown-field" => value["future_hint"] = true.into(), + _ => {} + } + let bytes = if kind == "corrupt" { + b"{broken".to_vec() + } else { + serde_json::to_vec(&value).expect("marker JSON") + }; + save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), bytes) + .await + .expect("persist rejected marker"); + let before = persisted_state(&store).await; + run_disabled_startup(CancellationToken::new(), store.clone()).await; + assert_eq!(persisted_state(&store).await, before, "{kind} must not become an automatic full rescan"); + } +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_rechecks_revision_after_waiting_for_leader_lock() { + let (_dir, store) = setup_scanner_cycle_store().await; + let mut marker = seed_cleanup(&store, "cleanup-pending").await; + let expected = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()) + .await + .expect("intent revision") + .1; + 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 leader ownership"); + let mut recovery = Box::pin(reset_scanner_cycle_recovery_for_intent( + CancellationToken::new(), + store.clone(), + Some(expected), + None, + )); + assert!(matches!(futures::poll!(&mut recovery), Poll::Pending)); + marker.state = "blocked".to_string(); + save_config( + store.clone(), + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + serde_json::to_vec(&marker).expect("replacement marker"), + ) + .await + .expect("replace intent while the fixture owns leader lock"); + let replaced = persisted_state(&store).await; + drop(guard); + let error = recovery.await.expect_err("old preflight cannot authorize replacement marker"); + assert!(error.to_string().contains("changed before recovery acquired ownership")); + assert_eq!(persisted_state(&store).await, replaced); +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_requires_phase_even_when_revision_matches() { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "blocked").await; + let before = persisted_state(&store).await; + let error = reset_scanner_cycle_recovery_for_intent(CancellationToken::new(), store.clone(), Some(before[1].1.clone()), None) + .await + .expect_err("a matching ETag alone is not operator cleanup authorization"); + assert!(error.to_string().contains("unchanged cleanup-pending")); + assert_eq!(persisted_state(&store).await, before); + reset_scanner_cycle_recovery(CancellationToken::new(), store.clone()) + .await + .expect("explicit v3 core retains full reset authorization"); + assert_reset_fences(&store).await; +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_lock_busy_preserves_intent_without_force_unlock() { + 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 error = resume_scanner_cycle_cleanup(CancellationToken::new(), store.clone()) + .await + .expect_err("busy leader must block recovery"); + assert!(error.to_string().contains("leader lock is busy")); + let status = scanner_cycle_recovery_status(); + assert_eq!(status.state, "cleanup-pending"); + assert!( + status + .reason + .as_deref() + .is_some_and(|reason| reason.contains("leader lock is busy")) + ); + assert!(!status.retryable, "disabled startup makes one attempt, not an automatic retry loop"); + assert!(!guard.is_lock_lost(), "recovery must not revoke the live owner"); + assert_eq!(persisted_state(&store).await, before); + drop(guard); + run_disabled_startup(CancellationToken::new(), store.clone()).await; + assert_reset_fences(&store).await; +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_movement_pause_preserves_intent_for_later_startup() { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "cleanup-pending").await; + let before = persisted_state(&store).await; + *store.rebalance_meta.write().await = Some(EcstoreRebalanceMeta { + id: "cleanup-movement".to_string(), + pool_stats: vec![EcstoreRebalanceStats { + participating: true, + info: EcstoreRebalanceInfo { + start_time: Some(time::OffsetDateTime::now_utc()), + status: EcstoreRebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }); + let error = resume_scanner_cycle_cleanup(CancellationToken::new(), store.clone()) + .await + .expect_err("movement must block reset publication"); + assert!(error.to_string().contains("blocked by data movement")); + let status = scanner_cycle_recovery_status(); + assert_eq!(status.state, "cleanup-pending"); + assert!( + status + .reason + .as_deref() + .is_some_and(|reason| reason.contains("blocked by data movement")) + ); + assert_eq!(persisted_state(&store).await, before); + *store.rebalance_meta.write().await = None; + run_disabled_startup(CancellationToken::new(), store.clone()).await; + assert_reset_fences(&store).await; +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_cancelled_startup_preserves_persisted_work() { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "cleanup-pending").await; + let before = persisted_state(&store).await; + let ctx = CancellationToken::new(); + ctx.cancel(); + run_disabled_startup(ctx, store.clone()).await; + assert_eq!(persisted_state(&store).await, before); +} + +#[tokio::test(start_paused = true)] +#[serial] +async fn disabled_cleanup_probe_obeys_cancellation_and_existing_io_deadline() { + for cancel in [false, true] { + let store = Arc::new(MemoryConfigStore::default()); + store.delayed_gets.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()), + data_usage_persist_timeout().saturating_add(Duration::from_secs(60)), + ); + let ctx = CancellationToken::new(); + let mut probe = Box::pin(read_scanner_cleanup_marker(store.clone(), &ctx)); + assert!(matches!(futures::poll!(&mut probe), Poll::Pending)); + if cancel { + ctx.cancel(); + } else { + advance(data_usage_persist_timeout()).await; + } + let error = probe.await.expect_err("pending read must be bounded"); + assert!(error.to_string().contains(if cancel { "cancelled" } else { "timed out" })); + assert!(store.put_counts.lock().await.is_empty(), "probe must remain read-only"); + } +} + +#[test] +#[serial] +fn disabled_cleanup_old_observation_cannot_overwrite_a_new_completion() { + let original = scanner_cycle_recovery_status(); + let healthy = ScannerCycleRecoveryStatus { + state: "healthy".to_string(), + ..Default::default() + }; + publish_scanner_cleanup_status(healthy.clone(), None).expect("first status version"); + let old = scanner_cleanup_status_version(); + publish_scanner_cleanup_status(healthy, None).expect("a newer completion may have identical fields"); + assert!( + publish_scanner_cleanup_status( + ScannerCycleRecoveryStatus { + state: "cleanup-pending".to_string(), + reason: Some("old lock wait failed".to_string()), + ..Default::default() + }, + Some(old) + ) + .is_none() + ); + assert_eq!(scanner_cycle_recovery_status().state, "healthy"); + publish_scanner_cleanup_status(original, None).expect("restore prior observation"); +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_owned_read_and_write_failures_keep_specific_status() { + for (stage, newer_completion) in [ + (cleanup_io_fault::Stage::PrimaryRead, false), + (cleanup_io_fault::Stage::PrimaryWrite, false), + (cleanup_io_fault::Stage::PrimaryWrite, true), + ] { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "cleanup-pending").await; + let before = persisted_state(&store).await; + let injection = cleanup_io_fault::install(&store, stage, newer_completion); + let error = resume_scanner_cycle_cleanup(CancellationToken::new(), store.clone()) + .await + .expect_err("injected owned I/O boundary"); + assert!( + injection.fired_while_owned(), + "fault must occur after real leader ownership and marker validation" + ); + let expected_error = match stage { + cleanup_io_fault::Stage::PrimaryRead => "injected primary read failure", + cleanup_io_fault::Stage::PrimaryWrite => "injected primary write failure", + cleanup_io_fault::Stage::UsageFence => "injected usage fence failure", + }; + assert!(error.to_string().contains(expected_error)); + let status = scanner_cycle_recovery_status(); + if newer_completion { + assert_eq!(status.state, "healthy", "old error cannot overwrite a newer completion observation"); + assert!(status.reason.is_none()); + } else { + assert_eq!(status.state, "cleanup-pending"); + assert!( + status.reason.as_deref().is_some_and(|reason| reason.contains(expected_error)), + "{status:?}" + ); + } + let after = persisted_state(&store).await; + assert_eq!(after[0], before[0], "failed primary I/O must preserve its prior revision"); + assert_eq!(after[2], before[2], "failed primary I/O must not advance the usage fence"); + let marker: ScannerCycleRecoveryMarker = + serde_json::from_slice(after[1].0.as_deref().expect("durable marker retained")).expect("valid cleanup marker"); + assert_eq!(marker.state, "cleanup-pending"); + drop(injection); + run_disabled_startup(CancellationToken::new(), store.clone()).await; + assert_reset_fences(&store).await; + } +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_invalidated_observation_never_becomes_unconditional() { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "cleanup-pending").await; + let revision = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()) + .await + .expect("marker revision") + .1; + let newer = ScannerCycleRecoveryStatus { + state: "healthy".to_string(), + reason: Some("newer completion owner".to_string()), + ..Default::default() + }; + publish_scanner_cleanup_status(newer.clone(), None).expect("newer observation"); + let mut invalidated = None; + reset_scanner_cycle_recovery_for_intent(CancellationToken::new(), store.clone(), Some(revision), Some(&mut invalidated)) + .await + .expect("metadata cleanup may complete without owning the newest status observation"); + assert!(invalidated.is_none()); + assert_eq!( + serde_json::to_value(scanner_cycle_recovery_status()).expect("observed status"), + serde_json::to_value(newer).expect("newer status") + ); + assert_reset_fences(&store).await; +} + +#[tokio::test] +#[serial] +async fn disabled_cleanup_later_failure_preserves_rebuilt_primary_status_identity() { + for newer_completion in [false, true] { + let (_dir, store) = setup_scanner_cycle_store().await; + seed_cleanup(&store, "cleanup-pending").await; + save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), b"corrupt-cycle".to_vec()) + .await + .expect("force the full reconstruction branch"); + let before = persisted_state(&store).await; + let injection = cleanup_io_fault::install(&store, cleanup_io_fault::Stage::UsageFence, newer_completion); + let error = resume_scanner_cycle_cleanup(CancellationToken::new(), store.clone()) + .await + .expect_err("fail after primary publication"); + assert!(injection.fired_while_owned()); + assert!(error.to_string().contains("injected usage fence failure")); + let after = persisted_state(&store).await; + assert_ne!(after[0].1, before[0].1, "the primary write must actually commit before this failure"); + let (cycle, epoch) = + decode_scanner_cycle_state(after[0].0.as_deref().expect("rebuilt primary")).expect("valid durable reconstruction"); + assert_eq!((cycle.current, cycle.next, epoch), (0, 42, 8)); + assert_eq!(after[2], before[2], "usage fence publication was rejected"); + let marker: ScannerCycleRecoveryMarker = + serde_json::from_slice(after[1].0.as_deref().expect("cleanup marker retained")).expect("valid cleanup marker"); + assert_eq!(marker.state, "cleanup-pending"); + let status = scanner_cycle_recovery_status(); + if newer_completion { + assert_eq!(status.state, "healthy"); + assert!( + status.reason.is_none(), + "old core and outer failure must both retain invalidated ownership" + ); + } else { + let DataUsageCacheRevision::Etag(etag) = &after[0].1 else { + panic!("rebuilt primary must have a revision"); + }; + assert_eq!(status.state, "cleanup-pending"); + assert_eq!(status.primary_revision.as_deref(), Some(etag.as_str())); + assert_eq!(status.generation, Some(cycle.next)); + assert_eq!(status.leader_epoch, Some(epoch)); + assert!( + status + .reason + .as_deref() + .is_some_and(|reason| reason.contains("injected usage fence failure")) + ); + } + } +} diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index c931f3302..7a06e9fa3 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -179,6 +179,14 @@ The reset does not delete metadata files by hand and does not publish an authori | data movement | wait for decommission or rebalance to leave the scanner metadata path, then retry | | invalid scanner cycle state | run `POST /v3/scanner/cycle-state/reset` with `{"mode":"full-rescan"}` first | +## Cleanup With The Scanner Disabled + +With `RUSTFS_SCANNER_ENABLED=false`, startup makes one controlled attempt to finish a previously persisted cycle reset whose validated recovery marker is already `cleanup-pending`. This is metadata cleanup only: it does not start the ordinary scanner loop, scan namespaces, accept a new reset request, or automatically perform a usage-state `full-rebuild`. Missing, merely `blocked`, unknown-version, unknown-phase, or corrupt markers do not authorize an automatic reset. + +The attempt uses the existing leader lock and revalidates the observed marker revision and phase after acquiring it. A busy leader or data-movement pause leaves the marker intact and is reported through `cycle_recovery.state` and `cycle_recovery.reason` in the existing scanner status response. There is no automatic retry loop while disabled. After resolving the blocker, explicitly retry `POST /v3/scanner/cycle-state/reset` with `{"mode":"full-rescan"}`, or restart to make another controlled attempt. The v3 reset routes remain synchronous and return their existing successful HTTP 200 responses; no asynchronous HTTP 202 acceptance is introduced. + +The startup probe is cancellation-aware and uses the existing cache persistence I/O timeout. Shutdown waits only for the existing server shutdown timeout. If the cleanup task cannot join in that window, the `scanner_cleanup_not_joined` warning means completion is unconfirmed, not drained. The task is not force-aborted or force-unlocked while its runtime remains alive; it retains its existing namespace/admission guards, and durable marker/fence state remains authoritative. Inspect status before retrying. This does not establish a hard deadline for an unresponsive storage operation or prove that I/O has drained when the process or runtime subsequently exits. Task-ownership timeout tests are not storage fsync, commit-tail, or process-crash durability evidence. + ## Data Movement Pauses RustFS uses a `global_pause` policy while pool decommission or rebalance can hide scanner metadata: usage publication, lifecycle discovery, tier cleanup discovery, scanner-originated heal and bitrot checks, and replication discovery are deferred together. A failed or canceled decommission remains a publication barrier until an operator retries or clears it. The same pause and estimate objects are included in `GET /v3/ilm/expiry/status`. diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs index 5586511c1..0b19881e0 100644 --- a/rustfs/src/startup_lifecycle.rs +++ b/rustfs/src/startup_lifecycle.rs @@ -15,7 +15,9 @@ use crate::storage_api::startup::lifecycle::ECStore; use crate::{ connect::runtime::shutdown_connect_runtimes, - server::{ServiceStateManager, ShutdownHandle, start_persisted_event_notifier_reconciler, wait_for_shutdown}, + server::{ + SHUTDOWN_TIMEOUT, ServiceStateManager, ShutdownHandle, start_persisted_event_notifier_reconciler, wait_for_shutdown, + }, startup_iam::{IamBootstrapDisposition, publish_ready_for_iam_bootstrap}, startup_runtime_sources, startup_services::StartupServiceRuntime, @@ -23,7 +25,7 @@ use crate::{ }; use rustfs_common::GlobalReadiness; use rustfs_object_capacity::capacity_manager::CapacityBackgroundTasks; -use rustfs_scanner::init_data_scanner; +use rustfs_scanner::init_scanner_with_recovery; use std::{ io::{Error, Result}, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, @@ -150,9 +152,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec startup_runtime_sources::publish_init_time_now().await; let event_notifier_reconciler = start_persisted_event_notifier_reconciler(store.clone(), shutdown_token.clone()); - if enable_scanner { - init_data_scanner(shutdown_token.clone(), store).await; - } + let scanner_cleanup = init_scanner_with_recovery(shutdown_token.clone(), store, enable_scanner).await; let shutdown_signal = wait_for_shutdown().await; run_startup_shutdown_sequence( @@ -166,6 +166,9 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec ) .await; shutdown_connect_runtimes(heartbeat, inventory).await; + if let Some(cleanup) = scanner_cleanup { + let _ = wait_for_scanner_cleanup(cleanup).await; + } if let Err(err) = event_notifier_reconciler.await { tracing::warn!( target: "rustfs::main::run", @@ -192,6 +195,40 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec Ok(()) } +async fn wait_for_scanner_cleanup(cleanup: tokio::task::JoinHandle<()>) -> bool { + // Dropping a JoinHandle detaches rather than aborts the task. Keep an + // in-flight reset's guards owned while this runtime remains alive. This + // bounded wait does not prove I/O drain at a later runtime/process exit. + match tokio::time::timeout(SHUTDOWN_TIMEOUT, cleanup).await { + Ok(Ok(())) => true, + Ok(Err(error)) => { + tracing::warn!( + target: "rustfs::main::run", + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "scanner_cleanup_join_failed", + reason = if error.is_cancelled() { "task_cancelled" } else { "task_panicked" }, + "Scanner cleanup task failed to join" + ); + false + } + Err(_) => { + tracing::warn!( + target: "rustfs::main::run", + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "scanner_cleanup_not_joined", + reason = "timeout", + timeout_secs = SHUTDOWN_TIMEOUT.as_secs(), + "Scanner cleanup completion is unconfirmed; task was not aborted" + ); + false + } + } +} + pub(crate) async fn publish_embedded_startup_ready( iam_bootstrap: IamBootstrapDisposition, readiness: &GlobalReadiness, @@ -226,12 +263,50 @@ pub(crate) fn log_embedded_server_ready(endpoint_address: SocketAddr) { #[cfg(test)] mod tests { - use super::{embedded_endpoint_address, mark_embedded_global_init_started}; + use super::{embedded_endpoint_address, mark_embedded_global_init_started, wait_for_scanner_cleanup}; use std::{ net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, sync::atomic::{AtomicBool, Ordering}, }; + #[tokio::test] + async fn disabled_cleanup_shutdown_joins_a_finished_task() { + assert!(wait_for_scanner_cleanup(tokio::spawn(async {})).await); + } + + #[tokio::test] + async fn disabled_cleanup_shutdown_reports_task_failure() { + assert!(!wait_for_scanner_cleanup(tokio::spawn(async { panic!("fixture cleanup failure") })).await); + } + + #[tokio::test(start_paused = true)] + async fn disabled_cleanup_shutdown_timeout_does_not_abort_owned_work() { + struct OwnedWork(std::sync::Arc); + impl Drop for OwnedWork { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + let dropped = std::sync::Arc::new(AtomicBool::new(false)); + let owned = OwnedWork(dropped.clone()); + let (started, ready) = tokio::sync::oneshot::channel(); + let (release, resume) = tokio::sync::oneshot::channel(); + let (finished, done) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(async move { + let owned = owned; + started.send(()).expect("work started"); + resume.await.expect("fixture releases owned work"); + drop(owned); + finished.send(()).expect("work completed"); + }); + ready.await.expect("task must actually be in flight"); + assert!(!wait_for_scanner_cleanup(task).await, "timeout is not a completed join"); + assert!(!dropped.load(Ordering::SeqCst), "timeout must not abort the reset's owned guards"); + release.send(()).expect("detached task remains alive"); + done.await.expect("owned work should finish after explicit release"); + assert!(dropped.load(Ordering::SeqCst)); + } + #[test] fn embedded_global_init_guard_allows_local_retry_before_mark() { let server_started = AtomicBool::new(false);