diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 75c3ca06e..b699697be 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -146,6 +146,11 @@ pub static LEGACY_DATA_USAGE_OBJ_NAME_PATH: LazyLock = pub static DATA_USAGE_BLOOM_NAME_PATH: LazyLock = LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{DATA_USAGE_BLOOM_NAME}")); +/// Durable companion object for a cycle-state object which cannot be decoded. +/// The primary object is deliberately never replaced or deleted by recovery. +pub static DATA_USAGE_BLOOM_RECOVERY_PATH: LazyLock = + LazyLock::new(|| format!("{}.recovery-required.json", DATA_USAGE_BLOOM_NAME_PATH.as_str())); + pub static BACKGROUND_HEAL_INFO_PATH: LazyLock = LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}.background-heal.json")); diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 0e347a8c1..164ee7214 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -75,7 +75,10 @@ pub use remote_scanner::{ }; pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_status, validate_scanner_runtime_config}; pub use rustfs_common::last_minute; -pub use scanner::{ScannerCycleScheduleStatus, init_data_scanner, scanner_cycle_schedule_status, scanner_topology_digest}; +pub use scanner::{ + ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, init_data_scanner, + reset_scanner_cycle_recovery, scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_topology_digest, +}; pub use scanner_io::{ ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state, diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 4db558a54..ce7b5ed13 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -54,9 +54,7 @@ use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELA use rustfs_data_usage::observed_data_usage_is_newer; use serde::{Deserialize, Serialize}; use sha2::{Digest as _, Sha256}; -#[cfg(test)] -use tokio::sync::Notify; -use tokio::sync::mpsc; +use tokio::sync::{Notify, mpsc}; use tokio::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; @@ -104,6 +102,13 @@ const CLEAN_IDLE_BACKOFF_FACTOR: u32 = 2; /// unavailable peer cannot drive a tight retry loop. const SCANNER_RETRY_BASE_INTERVAL: Duration = Duration::from_secs(5); const SCANNER_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30 * 60); +/// A transient backend outage remains self-healing after the short retry +/// budget is exhausted, but the probe is intentionally sparse until storage +/// recovers or an operator reset wakes the scanner. +const SCANNER_CYCLE_RECOVERY_PAUSED_INTERVAL: Duration = Duration::from_secs(5 * 60); +/// Permanent recovery states still get a sparse status probe so a reset that +/// races the wait registration cannot leave the scanner asleep forever. +const SCANNER_CYCLE_RECOVERY_BLOCKED_PROBE_INTERVAL: Duration = Duration::from_secs(5 * 60); const SCANNER_LEADER_LOCK_POLL_INTERVAL: Duration = Duration::from_secs(1); #[cfg(not(test))] const SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30); @@ -125,6 +130,12 @@ type ScannerCycleStatePersistTestHook = (u64, Arc); static SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK: LazyLock>> = LazyLock::new(|| StdMutex::new(None)); +static SCANNER_CYCLE_RECOVERY_WAKE: LazyLock = LazyLock::new(Notify::new); + +pub(super) fn notify_scanner_cycle_recovery_wake() { + SCANNER_CYCLE_RECOVERY_WAKE.notify_waiters(); +} + #[cfg(test)] struct ScannerCycleStatePersistTestHookGuard; @@ -576,19 +587,21 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { tokio::time::sleep(sleep_time).await; } + let mut transient_backoff = ScannerRetryBackoff::default(); + let mut recovery_retry_count = 0_u32; loop { if ctx_clone.is_cancelled() { break; } - if let Err(e) = run_data_scanner_with_maintenance_state( + let run_result = run_data_scanner_with_maintenance_state( ctx_clone.clone(), storeapi_clone.clone(), startup_features, startup_maintenance_generation, ) - .await - { + .await; + if let Err(e) = &run_result { error!( target: "rustfs::scanner", event = EVENT_SCANNER_CYCLE_STATE, @@ -599,11 +612,52 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { "Scanner runtime iteration failed" ); } + let recovery_status = scanner_cycle_recovery_status(); + if recovery_status.retryable { + recovery_retry_count = recovery_retry_count.saturating_add(1); + let _ = record_scanner_cycle_recovery_retry(recovery_retry_count); + } else { + recovery_retry_count = 0; + } + + let recovery_status = scanner_cycle_recovery_status(); + if recovery_status.state == "paused" { + transient_backoff.record_retryable_cycle(false); + tokio::select! { + _ = ctx_clone.cancelled() => break, + _ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {}, + _ = tokio::time::sleep(SCANNER_CYCLE_RECOVERY_PAUSED_INTERVAL) => {}, + } + recovery_retry_count = 0; + continue; + } + if !recovery_status.retryable + && matches!(recovery_status.state.as_str(), "blocked" | "recovery-required" | "cleanup-pending") + { + transient_backoff.record_retryable_cycle(false); + tokio::select! { + _ = ctx_clone.cancelled() => break, + _ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {}, + _ = tokio::time::sleep(SCANNER_CYCLE_RECOVERY_BLOCKED_PROBE_INTERVAL) => {}, + } + continue; + } + + let retry_delay = if recovery_status.retryable || run_result.is_err() { + transient_backoff.record_retryable_cycle(true); + transient_backoff + .retry_interval(scanner_cycle_interval()) + .unwrap_or(SCANNER_RETRY_BASE_INTERVAL) + } else { + transient_backoff.record_retryable_cycle(false); + randomized_cycle_delay() + }; // Backoff before retrying after lock contention or scanner-level failures. // Keep this cancellation-aware so shutdown is not delayed by backoff sleep. tokio::select! { _ = ctx_clone.cancelled() => break, - _ = tokio::time::sleep(randomized_cycle_delay()) => {} + _ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {}, + _ = tokio::time::sleep(retry_delay) => {} } } }); @@ -1606,40 +1660,22 @@ async fn run_data_scanner_with_maintenance_state( observe_scanner_activity(&storeapi, distributed, &mut scanner_activity_seen).await; } - let (buf, mut cycle_revision) = match read_config_with_revision(storeapi.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()).await { - Ok((buf, revision)) => (buf.unwrap_or_default(), revision), - Err(err) => { - error!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %&*DATA_USAGE_BLOOM_NAME_PATH, - state = "revision_load_failed", - error = %err, - "Scanner cycle state revision load failed" - ); - global_metrics().set_cycle(None).await; - return Ok(()); - } - }; - let (mut cycle_info, mut leader_epoch) = match decode_scanner_cycle_state_for_startup(&buf) { - Ok(state) => state, - Err(err) => { - error!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %&*DATA_USAGE_BLOOM_NAME_PATH, - state = "cycle_decode_failed", - error = %err, - "Scanner stopped because persisted cycle state is invalid" - ); - global_metrics().set_cycle(None).await; - return Ok(()); - } - }; + let (mut cycle_info, mut leader_epoch, mut cycle_revision) = + match load_scanner_cycle_state_for_startup(storeapi.clone()).await { + ScannerCycleStateStartup::Ready { + cycle, + leader_epoch, + revision, + } => (cycle, leader_epoch, revision), + ScannerCycleStateStartup::Blocked => { + global_metrics().set_cycle(None).await; + return Ok(()); + } + ScannerCycleStateStartup::Transient(err) => { + global_metrics().set_cycle(None).await; + return Err(err); + } + }; let usage_floor = match persisted_usage_floor(storeapi.clone()).await { Ok(floor) => floor, Err(err) => { @@ -2219,7 +2255,12 @@ pub(crate) use activity::{ pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance}; #[cfg(test)] pub(crate) use cycle_state::encode_scanner_cycle_fence_for_test; -pub(crate) use cycle_state::{current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence}; +pub use cycle_state::{ + ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, reset_scanner_cycle_recovery, scanner_cycle_recovery_status, +}; +pub(crate) use cycle_state::{ + current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence, load_scanner_cycle_state_for_startup, +}; pub use heal_info::{BackgroundHealInfo, read_background_heal_info, save_background_heal_info}; pub use usage_store::store_data_usage_in_backend; diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 6f3af4b81..fe6f6baf6 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -13,6 +13,952 @@ // limitations under the License. /// Scanner cycle-state codec, persisted usage floors, and cycle-state persistence. use super::*; +use crate::ScannerGetObjectReader; +use crate::data_usage_define::DATA_USAGE_BLOOM_RECOVERY_PATH; +use crate::storage_api::owner::ObjectIO as _; +use tokio::io::AsyncReadExt as _; + +const SCANNER_CYCLE_RECOVERY_SCHEMA_VERSION: u16 = 1; +const MAX_SCANNER_CYCLE_STATE_BYTES: u64 = 1024 * 1024; +pub(super) const MAX_SCANNER_CYCLE_RECOVERY_RETRIES: u32 = 5; +const METRIC_SCANNER_CYCLE_RECOVERY_REQUIRED: &str = "rustfs_scanner_cycle_recovery_required"; +const METRIC_SCANNER_CYCLE_RECOVERY_RETRY_COUNT: &str = "rustfs_scanner_cycle_recovery_retry_count"; + +#[derive(Clone, Debug, Default, Serialize)] +pub struct ScannerCycleRecoveryStatus { + /// The immutable primary object whose revision is being guarded. + pub path: String, + /// The companion marker/quarantine object containing the recovery evidence. + pub quarantine_path: Option, + pub state: String, + pub classification: Option, + pub primary_revision: Option, + pub generation: Option, + pub leader_epoch: Option, + pub first_detected_at_unix_secs: Option, + pub last_attempt_at_unix_secs: Option, + pub retry_count: u64, + pub max_retries: u32, + /// Whether the scanner may retry this state automatically. + pub retryable: bool, + pub reason: Option, +} + +static SCANNER_CYCLE_RECOVERY_STATUS: LazyLock> = LazyLock::new(|| { + RwLock::new(ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), + state: "healthy".to_string(), + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + ..Default::default() + }) +}); + +pub fn scanner_cycle_recovery_status() -> ScannerCycleRecoveryStatus { + SCANNER_CYCLE_RECOVERY_STATUS + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() +} + +fn set_scanner_cycle_recovery_status(status: ScannerCycleRecoveryStatus) { + let recovery_required = if matches!(status.state.as_str(), "blocked" | "paused" | "recovery-required" | "cleanup-pending") { + 1.0 + } else { + 0.0 + }; + 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; +} + +pub(super) fn record_scanner_cycle_recovery_retry(attempt: u32) -> bool { + let mut status = scanner_cycle_recovery_status(); + status.retry_count = u64::from(attempt); + status.last_attempt_at_unix_secs = Some(unix_now_secs()); + if attempt >= MAX_SCANNER_CYCLE_RECOVERY_RETRIES { + status.state = "paused".to_string(); + status.retryable = false; + status.reason = Some("scanner cycle recovery retry budget reached; sparse backend probes continue".to_string()); + set_scanner_cycle_recovery_status(status); + false + } else { + status.retryable = true; + set_scanner_cycle_recovery_status(status); + true + } +} + +fn unix_now_secs() -> u64 { + u64::try_from(Utc::now().timestamp()).unwrap_or(0) +} + +fn recovery_status(state: &str, reason: Option<&str>, retryable: bool) -> ScannerCycleRecoveryStatus { + ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), + state: state.to_string(), + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + retryable, + last_attempt_at_unix_secs: Some(unix_now_secs()), + reason: reason.map(str::to_string), + ..Default::default() + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ScannerCycleRecoveryMarker { + pub schema_version: u16, + pub primary_revision: String, + pub generation: u64, + pub leader_epoch: u64, + pub classification: String, + pub first_detected_at_unix_secs: u64, + pub last_attempt_at_unix_secs: u64, + pub retry_count: u64, + pub reason: String, + pub path: String, + pub quarantine_path: String, + /// `blocked` means the marker guards the primary revision; `cleanup-pending` + /// means an operator reset is in progress and must remain fenced across a + /// restart, even if the primary object is subsequently rewritten. + #[serde(default = "default_recovery_marker_state")] + pub state: String, +} + +fn default_recovery_marker_state() -> String { + "blocked".to_string() +} + +#[derive(Debug, Deserialize)] +struct ScannerCycleRecoveryMarkerCompat { + schema_version: Option, + primary_revision: Option, + classification: Option, + first_detected_at_unix_secs: Option, + last_attempt_at_unix_secs: Option, + retry_count: Option, + reason: Option, + path: Option, + quarantine_path: Option, + state: Option, +} + +#[derive(Debug)] +pub(crate) enum ScannerCycleStateStartup { + Ready { + cycle: CurrentCycle, + leader_epoch: u64, + revision: DataUsageCacheRevision, + }, + Blocked, + Transient(ScannerError), +} + +#[derive(Debug, thiserror::Error)] +enum CycleRecoveryMarkerReadError { + #[error("cycle recovery marker backend read failed: {0}")] + Backend(#[source] EcstoreError), + #[error("invalid cycle recovery marker: {0}")] + Invalid(&'static str), + #[error("cycle recovery marker revision changed while publishing")] + Conflict, +} + +#[derive(Debug, thiserror::Error)] +enum CycleStateBodyReadError { + #[error("scanner cycle state exceeds the bounded object size")] + TooLarge, + #[error("scanner cycle state body read failed: {0}")] + Backend(#[source] EcstoreError), +} + +fn recovery_status_from_marker(marker: &ScannerCycleRecoveryMarker, state: &str) -> ScannerCycleRecoveryStatus { + ScannerCycleRecoveryStatus { + path: marker.path.clone(), + quarantine_path: Some(marker.quarantine_path.clone()), + state: state.to_string(), + classification: Some(marker.classification.clone()), + primary_revision: Some(marker.primary_revision.clone()), + generation: Some(marker.generation), + leader_epoch: Some(marker.leader_epoch), + first_detected_at_unix_secs: Some(marker.first_detected_at_unix_secs), + last_attempt_at_unix_secs: Some(marker.last_attempt_at_unix_secs), + retry_count: marker.retry_count, + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + retryable: false, + reason: Some(marker.reason.clone()), + } +} + +fn marker_matches_revision(marker: &ScannerCycleRecoveryMarker, revision: &DataUsageCacheRevision) -> bool { + matches!(revision, DataUsageCacheRevision::Etag(etag) if marker.primary_revision == *etag) +} + +fn validate_recovery_marker(marker: &ScannerCycleRecoveryMarker) -> Result<(), &'static str> { + if marker.schema_version != SCANNER_CYCLE_RECOVERY_SCHEMA_VERSION { + return Err("cycle recovery marker schema is unsupported"); + } + if marker.primary_revision.is_empty() { + return Err("cycle recovery marker has no primary revision"); + } + if marker.path != *DATA_USAGE_BLOOM_NAME_PATH { + return Err("cycle recovery marker path does not match the scanner scope"); + } + if marker.quarantine_path != *DATA_USAGE_BLOOM_RECOVERY_PATH { + return Err("cycle recovery marker quarantine path does not match the scanner scope"); + } + if !matches!(marker.classification.as_str(), "corrupt" | "future_schema") { + return Err("cycle recovery marker classification is invalid"); + } + if !matches!(marker.state.as_str(), "blocked" | "cleanup-pending") { + return Err("cycle recovery marker state is invalid"); + } + Ok(()) +} + +/// Decode only the stable scope and revision fields needed by an authenticated +/// full-rescan reset. Startup keeps the strict decoder above so a newer marker +/// cannot be interpreted as a trusted cursor; reset deliberately rebuilds from +/// the persisted usage floor instead. +pub(super) fn decode_recovery_marker_for_reset( + data: &[u8], + marker_revision: &DataUsageCacheRevision, +) -> Result { + if !matches!(marker_revision, DataUsageCacheRevision::Etag(_)) { + return Err(ScannerError::Other("cycle recovery marker has no object revision".to_string())); + } + let compat = serde_json::from_slice::(data).ok(); + let _schema_version = compat.as_ref().and_then(|marker| marker.schema_version); + let primary_revision = compat + .as_ref() + .and_then(|marker| marker.primary_revision.clone()) + .filter(|revision| !revision.is_empty()) + .unwrap_or_default(); + let path = compat + .as_ref() + .and_then(|marker| marker.path.clone()) + .unwrap_or_else(|| DATA_USAGE_BLOOM_NAME_PATH.clone()); + let quarantine_path = compat + .as_ref() + .and_then(|marker| marker.quarantine_path.clone()) + .unwrap_or_else(|| DATA_USAGE_BLOOM_RECOVERY_PATH.clone()); + if path != *DATA_USAGE_BLOOM_NAME_PATH || quarantine_path != *DATA_USAGE_BLOOM_RECOVERY_PATH { + return Err(ScannerError::Other( + "cycle recovery marker path does not match the scanner scope".to_string(), + )); + } + let classification = match compat.as_ref().and_then(|marker| marker.classification.as_deref()) { + Some("corrupt") => "corrupt", + Some("future_schema") | None => "future_schema", + Some(_) => "future_schema", + }; + let state = match compat.as_ref().and_then(|marker| marker.state.as_deref()) { + Some("cleanup-pending") => "cleanup-pending", + _ => "blocked", + }; + let now = unix_now_secs(); + Ok(ScannerCycleRecoveryMarker { + schema_version: SCANNER_CYCLE_RECOVERY_SCHEMA_VERSION, + primary_revision, + // Cursor and epoch values from an unknown marker are audit-only data; + // the reset path intentionally rebuilds both from the verified usage + // floor instead of carrying them across a version boundary. + generation: 0, + leader_epoch: 0, + classification: classification.to_string(), + first_detected_at_unix_secs: compat + .as_ref() + .and_then(|marker| marker.first_detected_at_unix_secs) + .unwrap_or(now), + last_attempt_at_unix_secs: compat + .as_ref() + .and_then(|marker| marker.last_attempt_at_unix_secs) + .unwrap_or(now), + retry_count: compat.as_ref().and_then(|marker| marker.retry_count).unwrap_or(0), + reason: compat + .as_ref() + .and_then(|marker| marker.reason.clone()) + .unwrap_or_else(|| "operator requested full scanner rescan".to_string()), + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(), + state: state.to_string(), + }) +} + +async fn read_cycle_state_body(reader: &mut ScannerGetObjectReader) -> Result, CycleStateBodyReadError> { + let max_len = usize::try_from(MAX_SCANNER_CYCLE_STATE_BYTES).unwrap_or(usize::MAX); + let mut data = Vec::new(); + reader + .take(MAX_SCANNER_CYCLE_STATE_BYTES.saturating_add(1)) + .read_to_end(&mut data) + .await + .map_err(|err| CycleStateBodyReadError::Backend(EcstoreError::other(err)))?; + if data.len() > max_len { + return Err(CycleStateBodyReadError::TooLarge); + } + Ok(data) +} + +fn cycle_state_classification(buf: &[u8]) -> (&'static str, &'static str) { + if buf.len() >= 16 && &buf[8..12] == b"RSCY" && &buf[8..16] != SCANNER_CYCLE_STATE_MAGIC { + ("future_schema", "scanner cycle state schema is newer than this reader") + } else { + ("corrupt", "scanner cycle state failed validation") + } +} + +fn cycle_state_generation_and_epoch(buf: &[u8]) -> (u64, u64) { + let generation = buf + .get(..8) + .and_then(|bytes| bytes.try_into().ok()) + .map(u64::from_le_bytes) + .unwrap_or(0); + let leader_epoch = if buf.len() >= SCANNER_CYCLE_STATE_HEADER_LEN && &buf[8..16] == SCANNER_CYCLE_STATE_MAGIC { + u64::from_le_bytes(buf[16..24].try_into().unwrap_or([0; 8])) + } else { + 0 + }; + (generation, leader_epoch) +} + +async fn persist_cycle_recovery_marker( + storeapi: Arc, + primary_revision: &DataUsageCacheRevision, + generation: u64, + leader_epoch: u64, + classification: &'static str, + reason: &'static str, +) -> Result { + let now = unix_now_secs(); + let (existing, existing_revision) = match read_cycle_recovery_marker_bytes(storeapi.clone()).await { + Ok(result) => result, + Err(err) => return Err(err), + }; + let existing_marker = existing + .as_deref() + .and_then(|bytes| serde_json::from_slice::(bytes).ok()); + let primary_revision = match primary_revision { + DataUsageCacheRevision::Etag(etag) => etag.clone(), + DataUsageCacheRevision::Missing => { + return Err(CycleRecoveryMarkerReadError::Invalid("cycle state recovery requires a primary revision")); + } + }; + let marker = ScannerCycleRecoveryMarker { + schema_version: SCANNER_CYCLE_RECOVERY_SCHEMA_VERSION, + primary_revision: primary_revision.clone(), + generation, + leader_epoch, + classification: classification.to_string(), + first_detected_at_unix_secs: existing_marker + .as_ref() + .filter(|marker| marker.primary_revision == primary_revision) + .map(|marker| marker.first_detected_at_unix_secs) + .unwrap_or(now), + last_attempt_at_unix_secs: now, + retry_count: existing_marker + .as_ref() + .filter(|marker| marker.primary_revision == primary_revision) + .map(|marker| marker.retry_count.saturating_add(1)) + .unwrap_or(0), + reason: reason.to_string(), + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(), + state: "blocked".to_string(), + }; + let bytes = serde_json::to_vec(&marker).map_err(|_| CycleRecoveryMarkerReadError::Invalid("marker serialization failed"))?; + let save_result = save_config_with_preconditions( + storeapi.clone(), + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + bytes, + existing_revision.preconditions(), + ) + .await; + match save_result { + Ok(_) => Ok(marker), + Err(EcstoreError::PreconditionFailed) => Err(CycleRecoveryMarkerReadError::Conflict), + Err(err) => Err(CycleRecoveryMarkerReadError::Backend(err)), + } +} + +async fn read_cycle_recovery_marker_bytes( + storeapi: Arc, +) -> Result<(Option>, DataUsageCacheRevision), CycleRecoveryMarkerReadError> { + let mut reader = match storeapi + .get_object_reader( + RUSTFS_META_BUCKET, + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + None, + http::HeaderMap::new(), + &ScannerObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(reader) => reader, + Err( + EcstoreError::FileNotFound + | EcstoreError::VolumeNotFound + | EcstoreError::ObjectNotFound(_, _) + | EcstoreError::BucketNotFound(_) + | EcstoreError::ConfigNotFound, + ) => { + return Ok((None, DataUsageCacheRevision::Missing)); + } + Err(err) => return Err(CycleRecoveryMarkerReadError::Backend(err)), + }; + let revision = reader + .object_info + .etag + .as_ref() + .filter(|etag| !etag.is_empty()) + .cloned() + .map(DataUsageCacheRevision::Etag) + .ok_or(CycleRecoveryMarkerReadError::Invalid("marker has no revision"))?; + if reader.object_info.is_dir || reader.object_info.size < 0 || reader.object_info.size > 64 * 1024 { + return Err(CycleRecoveryMarkerReadError::Invalid("marker exceeds the bounded object size")); + } + let mut data = Vec::new(); + (&mut reader) + .take(64 * 1024 + 1) + .read_to_end(&mut data) + .await + .map_err(|err| CycleRecoveryMarkerReadError::Backend(EcstoreError::other(err)))?; + if data.len() > 64 * 1024 { + return Err(CycleRecoveryMarkerReadError::Invalid("marker exceeds the bounded object size")); + } + if data.is_empty() { + return Err(CycleRecoveryMarkerReadError::Invalid("marker is empty")); + } + Ok((Some(data), revision)) +} + +async fn quarantine_invalid_cycle_state( + storeapi: Arc, + revision: &DataUsageCacheRevision, + buf: &[u8], +) -> ScannerCycleStateStartup { + let (classification, reason) = cycle_state_classification(buf); + let (generation, leader_epoch) = cycle_state_generation_and_epoch(buf); + quarantine_invalid_cycle_state_with_reason(storeapi, revision, generation, leader_epoch, classification, reason).await +} + +async fn quarantine_invalid_cycle_state_with_reason( + storeapi: Arc, + revision: &DataUsageCacheRevision, + generation: u64, + leader_epoch: u64, + classification: &'static str, + reason: &'static str, +) -> ScannerCycleStateStartup { + let now = unix_now_secs(); + let base_status = ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), + state: "recovery-required".to_string(), + classification: Some(classification.to_string()), + primary_revision: match revision { + DataUsageCacheRevision::Etag(etag) => Some(etag.clone()), + DataUsageCacheRevision::Missing => None, + }, + generation: Some(generation), + leader_epoch: Some(leader_epoch), + first_detected_at_unix_secs: Some(now), + last_attempt_at_unix_secs: Some(now), + retry_count: 0, + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + retryable: true, + reason: Some(reason.to_string()), + }; + set_scanner_cycle_recovery_status(base_status); + match persist_cycle_recovery_marker(storeapi, revision, generation, leader_epoch, classification, reason).await { + Ok(marker) => set_scanner_cycle_recovery_status(recovery_status_from_marker(&marker, "blocked")), + Err(CycleRecoveryMarkerReadError::Backend(_)) => { + // Keep the poison object untouched and retry marker creation with the + // bounded startup backoff; recovery-required never becomes healthy. + return ScannerCycleStateStartup::Transient(ScannerError::Other( + "failed to persist scanner cycle recovery marker".to_string(), + )); + } + Err(CycleRecoveryMarkerReadError::Conflict) => { + set_scanner_cycle_recovery_status(recovery_status( + "transient", + Some("cycle recovery marker revision changed while publishing"), + true, + )); + return ScannerCycleStateStartup::Transient(ScannerError::Other( + "cycle recovery marker revision changed while publishing".to_string(), + )); + } + Err(CycleRecoveryMarkerReadError::Invalid(reason)) => { + set_scanner_cycle_recovery_status(recovery_status("recovery-required", Some(reason), false)); + return ScannerCycleStateStartup::Blocked; + } + } + ScannerCycleStateStartup::Blocked +} + +async fn mark_cycle_recovery_cleanup_pending( + storeapi: Arc, + mut marker: ScannerCycleRecoveryMarker, + marker_revision: &DataUsageCacheRevision, +) -> Result<(ScannerCycleRecoveryMarker, DataUsageCacheRevision), ScannerError> { + marker.state = "cleanup-pending".to_string(); + marker.last_attempt_at_unix_secs = unix_now_secs(); + let bytes = serde_json::to_vec(&marker) + .map_err(|err| ScannerError::Other(format!("failed to encode cycle recovery marker: {err}")))?; + let info = save_config_with_preconditions( + storeapi.clone(), + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + bytes, + marker_revision.preconditions(), + ) + .await + .map_err(|err| ScannerError::Other(format!("failed to mark cycle recovery cleanup pending: {err}")))?; + let revision = info + .etag + .filter(|etag| !etag.is_empty()) + .map(DataUsageCacheRevision::Etag) + .ok_or_else(|| ScannerError::Other("cycle recovery marker save returned no revision".to_string()))?; + Ok((marker, revision)) +} + +pub(crate) async fn load_scanner_cycle_state_for_startup(storeapi: Arc) -> ScannerCycleStateStartup { + let marker = match read_cycle_recovery_marker_bytes(storeapi.clone()).await { + Ok((None, _)) => None, + Ok((Some(data), marker_revision)) => match serde_json::from_slice::(&data) { + Ok(marker) => match validate_recovery_marker(&marker) { + Ok(()) => Some((marker, marker_revision)), + Err(reason) => { + set_scanner_cycle_recovery_status(recovery_status("recovery-required", Some(reason), false)); + return ScannerCycleStateStartup::Blocked; + } + }, + Err(_) => { + set_scanner_cycle_recovery_status(recovery_status( + "recovery-required", + Some("cycle recovery marker is invalid"), + false, + )); + return ScannerCycleStateStartup::Blocked; + } + }, + Err(CycleRecoveryMarkerReadError::Backend(err)) => { + let status = recovery_status("transient", Some("cycle recovery marker I/O is temporarily unavailable"), true); + set_scanner_cycle_recovery_status(status); + return ScannerCycleStateStartup::Transient(ScannerError::Other(format!( + "failed to read scanner cycle recovery marker: {err}" + ))); + } + Err(CycleRecoveryMarkerReadError::Invalid(reason)) => { + set_scanner_cycle_recovery_status(recovery_status("recovery-required", Some(reason), false)); + return ScannerCycleStateStartup::Blocked; + } + Err(CycleRecoveryMarkerReadError::Conflict) => { + set_scanner_cycle_recovery_status(recovery_status( + "transient", + Some("cycle recovery marker revision changed while being inspected"), + true, + )); + return ScannerCycleStateStartup::Transient(ScannerError::Other( + "cycle recovery marker revision changed while being inspected".to_string(), + )); + } + }; + + let mut reader = match storeapi + .get_object_reader( + RUSTFS_META_BUCKET, + DATA_USAGE_BLOOM_NAME_PATH.as_str(), + None, + http::HeaderMap::new(), + &ScannerObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(reader) => reader, + Err( + EcstoreError::FileNotFound + | EcstoreError::VolumeNotFound + | EcstoreError::ObjectNotFound(_, _) + | EcstoreError::BucketNotFound(_) + | EcstoreError::ConfigNotFound, + ) => { + if let Some((marker, _)) = marker { + let state = if marker.state == "cleanup-pending" { + "cleanup-pending" + } else { + "recovery-required" + }; + set_scanner_cycle_recovery_status(recovery_status_from_marker(&marker, state)); + return ScannerCycleStateStartup::Blocked; + } + set_scanner_cycle_recovery_status(recovery_status("healthy", None, false)); + return ScannerCycleStateStartup::Ready { + cycle: CurrentCycle::default(), + leader_epoch: 0, + revision: DataUsageCacheRevision::Missing, + }; + } + Err(err) => { + set_scanner_cycle_recovery_status(recovery_status("transient", Some("cycle state could not be inspected"), true)); + return ScannerCycleStateStartup::Transient(ScannerError::Other(format!( + "failed to inspect scanner cycle state: {err}" + ))); + } + }; + let revision = reader + .object_info + .etag + .as_ref() + .filter(|etag| !etag.is_empty()) + .cloned() + .map(DataUsageCacheRevision::Etag); + let Some(revision) = revision else { + set_scanner_cycle_recovery_status(recovery_status("recovery-required", Some("cycle state has no revision"), false)); + return ScannerCycleStateStartup::Blocked; + }; + let max_size = i64::try_from(MAX_SCANNER_CYCLE_STATE_BYTES).unwrap_or(i64::MAX); + if reader.object_info.is_dir || reader.object_info.size < 0 || reader.object_info.size > max_size { + return quarantine_invalid_cycle_state_with_reason( + storeapi, + &revision, + 0, + 0, + "corrupt", + "scanner cycle state object is oversized or not a regular object", + ) + .await; + } + if let Some((marker, _)) = marker + .as_ref() + .filter(|(marker, _)| marker.state == "cleanup-pending" || marker_matches_revision(marker, &revision)) + { + let state = if marker.state == "cleanup-pending" { + "cleanup-pending" + } else { + "blocked" + }; + set_scanner_cycle_recovery_status(recovery_status_from_marker(marker, state)); + return ScannerCycleStateStartup::Blocked; + } + let data = match read_cycle_state_body(&mut reader).await { + Ok(data) => data, + Err(CycleStateBodyReadError::TooLarge) => { + return quarantine_invalid_cycle_state_with_reason( + storeapi, + &revision, + 0, + 0, + "corrupt", + "scanner cycle state exceeds the bounded object size", + ) + .await; + } + Err(CycleStateBodyReadError::Backend(err)) => { + set_scanner_cycle_recovery_status(recovery_status("transient", Some("cycle state read failed"), true)); + return ScannerCycleStateStartup::Transient(ScannerError::Other(format!( + "failed to read scanner cycle state: {err}" + ))); + } + }; + if data.is_empty() { + return quarantine_invalid_cycle_state_with_reason( + storeapi, + &revision, + 0, + 0, + "corrupt", + "scanner cycle state object is empty", + ) + .await; + } + match decode_scanner_cycle_state_for_startup(&data) { + Ok((cycle, leader_epoch)) => { + set_scanner_cycle_recovery_status(recovery_status("healthy", None, false)); + ScannerCycleStateStartup::Ready { + cycle, + leader_epoch, + revision, + } + } + Err(_) => quarantine_invalid_cycle_state(storeapi, &revision, &data).await, + } +} + +/// 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. +pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc) -> Result<(), ScannerError> { + let lock = storeapi + .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock") + .await + .map_err(|err| ScannerError::Other(format!("failed to acquire scanner leader lock: {err}")))?; + let guard = lock + .get_write_lock_quiet(Duration::from_secs(5)) + .await + .map_err(|err| ScannerError::Other(format!("scanner leader lock is busy: {err}")))?; + + if guard.is_lock_lost() { + return Err(ScannerError::Other("scanner leader lock was lost before recovery reset".to_string())); + } + + let (marker_data, marker_revision) = read_cycle_recovery_marker_bytes(storeapi.clone()) + .await + .map_err(|err| ScannerError::Other(format!("failed to read cycle recovery marker: {err}")))?; + let marker_data = marker_data.ok_or_else(|| ScannerError::Other("scanner cycle recovery marker is absent".to_string()))?; + let (marker, force_full_rescan) = match serde_json::from_slice::(&marker_data) { + Ok(marker) if validate_recovery_marker(&marker).is_ok() => (marker, false), + _ => (decode_recovery_marker_for_reset(&marker_data, &marker_revision)?, true), + }; + + if guard.is_lock_lost() { + return Err(ScannerError::Other( + "scanner leader lock was lost while reading recovery state".to_string(), + )); + } + + let (mut primary_reader, primary_revision) = match storeapi + .get_object_reader( + RUSTFS_META_BUCKET, + DATA_USAGE_BLOOM_NAME_PATH.as_str(), + None, + http::HeaderMap::new(), + &ScannerObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(reader) => { + let revision = reader + .object_info + .etag + .as_ref() + .filter(|etag| !etag.is_empty()) + .cloned() + .ok_or_else(|| ScannerError::Other("scanner cycle state has no revision".to_string()))?; + (Some(reader), DataUsageCacheRevision::Etag(revision)) + } + Err( + EcstoreError::FileNotFound + | EcstoreError::VolumeNotFound + | EcstoreError::ObjectNotFound(_, _) + | EcstoreError::BucketNotFound(_) + | EcstoreError::ConfigNotFound, + ) => (None, DataUsageCacheRevision::Missing), + Err(err) => return Err(ScannerError::Other(format!("failed to inspect scanner cycle state: {err}"))), + }; + let marker_guards_primary = + force_full_rescan || marker.state == "cleanup-pending" || marker_matches_revision(&marker, &primary_revision); + if !marker_guards_primary && let Some(mut reader) = primary_reader.take() { + // A newer, independently fenced primary is authoritative. A + // full-rescan reset must not overwrite that progress; it only + // removes the stale recovery marker after validating the state. + let max_size = i64::try_from(MAX_SCANNER_CYCLE_STATE_BYTES).unwrap_or(i64::MAX); + if reader.object_info.is_dir || reader.object_info.size < 0 || reader.object_info.size > max_size { + return Err(ScannerError::Other("scanner cycle state changed since recovery was recorded".to_string())); + } + let data = read_cycle_state_body(&mut reader) + .await + .map_err(|err| ScannerError::Other(format!("scanner cycle state changed since recovery was recorded: {err}")))?; + if data.is_empty() || decode_scanner_cycle_state_for_startup(&data).is_err() { + return Err(ScannerError::Other("scanner cycle state changed since recovery was recorded".to_string())); + } + storeapi + .delete_config_object( + RUSTFS_META_BUCKET, + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + ScannerObjectOptions { + delete_prefix: true, + delete_prefix_object: true, + no_lock: true, + http_preconditions: Some(marker_revision.preconditions()), + ..Default::default() + }, + ) + .await + .map_err(|err| ScannerError::Other(format!("failed to clear stale cycle recovery marker: {err}")))?; + set_scanner_cycle_recovery_status(recovery_status("healthy", None, false)); + super::notify_scanner_cycle_recovery_wake(); + return Ok(()); + } + + if guard.is_lock_lost() { + return Err(ScannerError::Other( + "scanner leader lock was lost before rebuilding cycle state".to_string(), + )); + } + + let floor = persisted_usage_floor(storeapi.clone()).await?; + // A full rescan must not trust a cursor recovered from a corrupt, future, + // or mixed-version marker. The durable usage floor is the only verified + // starting point; marker generation/epoch fields remain audit evidence. + let next = floor.next_cycle; + if next == u64::MAX { + return Err(ScannerError::Other("scanner cycle counter is exhausted".to_string())); + } + let leader_epoch = floor + .leader_epoch + .checked_add(1) + .ok_or_else(|| ScannerError::Other("scanner leader epoch is exhausted".to_string()))?; + let cycle = CurrentCycle { + next, + ..Default::default() + }; + let data = encode_scanner_cycle_state(&cycle, leader_epoch) + .map_err(|err| ScannerError::Other(format!("failed to encode rebuilt scanner cycle state: {err}")))?; + // Persist the cleanup-pending phase before rewriting the primary. If the + // process dies after the rewrite, startup still sees a durable fence and + // cannot mistake the partially completed reset for a healthy state. + let (marker, marker_revision) = if marker.state == "cleanup-pending" { + (marker, marker_revision) + } else { + mark_cycle_recovery_cleanup_pending(storeapi.clone(), marker, &marker_revision).await? + }; + let rebuilt_info = save_config_with_preconditions( + storeapi.clone(), + DATA_USAGE_BLOOM_NAME_PATH.as_str(), + data, + primary_revision.preconditions(), + ) + .await + .map_err(|err| ScannerError::Other(format!("failed to persist rebuilt scanner cycle state: {err}")))?; + let rebuilt_revision = rebuilt_info + .etag + .filter(|etag| !etag.is_empty()) + .ok_or_else(|| ScannerError::Other("rebuilt scanner cycle state has no revision".to_string()))?; + if guard.is_lock_lost() { + return Err(ScannerError::Other( + "scanner leader lock was lost after rebuilding cycle state".to_string(), + )); + } + if let Err(err) = fence_scanner_usage_epoch(&ctx, storeapi.clone(), leader_epoch).await { + set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), + state: "cleanup-pending".to_string(), + classification: Some(marker.classification.clone()), + primary_revision: Some(rebuilt_revision.clone()), + generation: Some(next), + leader_epoch: Some(leader_epoch), + first_detected_at_unix_secs: Some(marker.first_detected_at_unix_secs), + last_attempt_at_unix_secs: Some(unix_now_secs()), + retry_count: marker.retry_count, + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + retryable: false, + reason: Some("cycle state rebuilt but usage epoch fencing failed".to_string()), + }); + return Err(err); + } + + let current_revision = storeapi + .get_object_reader( + RUSTFS_META_BUCKET, + DATA_USAGE_BLOOM_NAME_PATH.as_str(), + None, + http::HeaderMap::new(), + &ScannerObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .map_err(|err| ScannerError::Other(format!("failed to verify rebuilt scanner cycle state: {err}")))? + .object_info + .etag + .filter(|etag| !etag.is_empty()) + .ok_or_else(|| ScannerError::Other("rebuilt scanner cycle state lost its revision".to_string()))?; + if current_revision != rebuilt_revision { + set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), + state: "cleanup-pending".to_string(), + classification: Some(marker.classification.clone()), + primary_revision: Some(current_revision), + generation: Some(next), + leader_epoch: Some(leader_epoch), + first_detected_at_unix_secs: Some(marker.first_detected_at_unix_secs), + last_attempt_at_unix_secs: Some(unix_now_secs()), + retry_count: marker.retry_count, + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + retryable: false, + reason: Some("rebuilt scanner cycle state changed before marker cleanup".to_string()), + }); + return Err(ScannerError::Other( + "rebuilt scanner cycle state changed before recovery marker cleanup".to_string(), + )); + } + + if guard.is_lock_lost() { + set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), + state: "cleanup-pending".to_string(), + classification: Some(marker.classification.clone()), + primary_revision: Some(rebuilt_revision.clone()), + generation: Some(next), + leader_epoch: Some(leader_epoch), + retry_count: marker.retry_count, + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + retryable: false, + reason: Some("cycle state rebuilt but recovery marker was not cleared".to_string()), + ..Default::default() + }); + return Err(ScannerError::Other( + "scanner leader lock was lost before clearing recovery marker".to_string(), + )); + } + + if let Err(err) = storeapi + .delete_config_object( + RUSTFS_META_BUCKET, + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + ScannerObjectOptions { + delete_prefix: true, + delete_prefix_object: true, + no_lock: true, + http_preconditions: Some(marker_revision.preconditions()), + ..Default::default() + }, + ) + .await + { + set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), + state: "cleanup-pending".to_string(), + classification: Some(marker.classification.clone()), + primary_revision: Some(rebuilt_revision.clone()), + generation: Some(next), + leader_epoch: Some(leader_epoch), + retry_count: marker.retry_count, + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + retryable: false, + reason: Some("cycle state rebuilt but recovery marker cleanup failed".to_string()), + ..Default::default() + }); + return Err(ScannerError::Other(format!("failed to clear cycle recovery marker: {err}"))); + } + set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus { + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: Some(DATA_USAGE_BLOOM_RECOVERY_PATH.clone()), + state: "healthy".to_string(), + max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES, + ..Default::default() + }); + super::notify_scanner_cycle_recovery_wake(); + Ok(()) +} #[derive(Debug, thiserror::Error)] pub(super) enum ScannerCycleStateError { diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index fd321bfc3..b381f9cc3 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -15,12 +15,13 @@ use super::*; use crate::EcstoreResult; use crate::{ - Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerGetObjectReader as GetObjectReader, - ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader, - init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx, + DATA_USAGE_BLOOM_RECOVERY_PATH, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, + ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, + ScannerPutObjReader as PutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, + init_local_disks_with_instance_ctx, }; use serial_test::serial; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::io::Cursor; use std::task::Poll; use temp_env::{with_var, with_var_unset}; @@ -152,6 +153,7 @@ impl Drop for ScannerDefaultCycleGuard { struct MemoryConfigStore { objects: Mutex>>, revisions: Mutex>, + non_regular_objects: Mutex>, fail_put_number: Mutex>, object_not_found_put_number: Mutex>, error_after_commit_put_number: Mutex>, @@ -192,12 +194,16 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore { .get(&key) .cloned() .ok_or(EcstoreError::FileNotFound)?; - let revision = *self.revisions.lock().await.entry(key).or_insert(1); + let data_len = i64::try_from(data.len()).expect("memory test object length should fit in i64"); + let revision = *self.revisions.lock().await.entry(key.clone()).or_insert(1); + let is_dir = self.non_regular_objects.lock().await.contains(&key); Ok(GetObjectReader { stream: Box::new(Cursor::new(data)), object_info: ObjectInfo { etag: Some(format!("memory-{revision}")), + size: data_len, + is_dir, ..Default::default() }, buffered_body: None, @@ -837,6 +843,327 @@ fn scanner_startup_fails_closed_on_nonempty_corrupt_cycle_state() { assert!(encode_scanner_cycle_state(&exhausted, 7).is_err()); } +#[tokio::test] +#[serial] +async fn corrupt_cycle_state_is_quarantined_once() { + let store = Arc::new(MemoryConfigStore::default()); + let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + store.objects.lock().await.insert(state_key.clone(), vec![1]); + store.revisions.lock().await.insert(state_key.clone(), 7); + + assert!(matches!( + load_scanner_cycle_state_for_startup(store.clone()).await, + ScannerCycleStateStartup::Blocked + )); + let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()); + let marker_data = store + .objects + .lock() + .await + .get(&marker_key) + .cloned() + .expect("corrupt state must leave a durable recovery marker"); + let marker: ScannerCycleRecoveryMarker = serde_json::from_slice(&marker_data).expect("marker should be valid JSON"); + assert_eq!(marker.primary_revision, "memory-7"); + assert_eq!(marker.path, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + assert_eq!(marker.quarantine_path, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()); + assert_eq!(marker.classification, "corrupt"); + + // A second startup sees the matching marker before consuming the poison body. + assert!(matches!( + load_scanner_cycle_state_for_startup(store.clone()).await, + ScannerCycleStateStartup::Blocked + )); + + // Replacing the primary object advances its revision; the stale marker must + // not quarantine the newer, valid state. + let cycle = CurrentCycle { + next: 9, + ..Default::default() + }; + let encoded = encode_scanner_cycle_state(&cycle, 3).expect("valid state should encode"); + store.objects.lock().await.insert(state_key.clone(), encoded); + store.revisions.lock().await.insert(state_key, 8); + assert!(matches!( + load_scanner_cycle_state_for_startup(store).await, + ScannerCycleStateStartup::Ready { + cycle: CurrentCycle { next: 9, .. }, + leader_epoch: 3, + .. + } + )); +} + +#[tokio::test] +#[serial] +async fn empty_cycle_state_object_is_quarantined_as_corrupt() { + let store = Arc::new(MemoryConfigStore::default()); + let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + store.objects.lock().await.insert(state_key.clone(), Vec::new()); + store.revisions.lock().await.insert(state_key, 6); + + assert!(matches!( + load_scanner_cycle_state_for_startup(store).await, + ScannerCycleStateStartup::Blocked + )); + assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("corrupt")); + assert!( + scanner_cycle_recovery_status() + .reason + .as_deref() + .is_some_and(|reason| reason.contains("empty")) + ); +} + +#[tokio::test] +#[serial] +async fn future_cycle_state_schema_is_recovery_required() { + let store = Arc::new(MemoryConfigStore::default()); + let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + let mut future = 17_u64.to_le_bytes().to_vec(); + future.extend_from_slice(b"RSCYC999"); + future.extend_from_slice(&4_u64.to_le_bytes()); + future.extend_from_slice(&[0x90]); + store.objects.lock().await.insert(state_key.clone(), future); + store.revisions.lock().await.insert(state_key, 13); + + assert!(matches!( + load_scanner_cycle_state_for_startup(store).await, + ScannerCycleStateStartup::Blocked + )); + assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("future_schema")); +} + +#[tokio::test] +#[serial] +async fn concurrent_leaders_cannot_quarantine_newer_cycle_state() { + let store = Arc::new(MemoryConfigStore::default()); + let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + store.objects.lock().await.insert(state_key.clone(), vec![1]); + store.revisions.lock().await.insert(state_key, 4); + + let (first, second) = tokio::join!( + load_scanner_cycle_state_for_startup(store.clone()), + load_scanner_cycle_state_for_startup(store.clone()), + ); + assert!(matches!(first, ScannerCycleStateStartup::Blocked)); + assert!(matches!(second, ScannerCycleStateStartup::Blocked)); + + let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()); + let marker_data = store + .objects + .lock() + .await + .get(&marker_key) + .cloned() + .expect("one contender must publish the recovery marker"); + let marker: ScannerCycleRecoveryMarker = serde_json::from_slice(&marker_data).expect("marker should decode"); + assert_eq!(marker.primary_revision, "memory-4"); +} + +#[tokio::test] +#[serial] +async fn cleanup_pending_marker_blocks_a_rewritten_primary_after_restart() { + let store = Arc::new(MemoryConfigStore::default()); + let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()); + let encoded = encode_scanner_cycle_state( + &CurrentCycle { + next: 12, + ..Default::default() + }, + 8, + ) + .expect("valid state should encode"); + store.objects.lock().await.insert(state_key.clone(), encoded); + store.revisions.lock().await.insert(state_key, 22); + let marker = ScannerCycleRecoveryMarker { + schema_version: 1, + primary_revision: "memory-21".to_string(), + generation: 11, + leader_epoch: 7, + classification: "corrupt".to_string(), + first_detected_at_unix_secs: 1, + last_attempt_at_unix_secs: 2, + retry_count: 1, + reason: "reset in progress".to_string(), + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(), + state: "cleanup-pending".to_string(), + }; + store + .objects + .lock() + .await + .insert(marker_key.clone(), serde_json::to_vec(&marker).expect("marker should encode")); + store.revisions.lock().await.insert(marker_key, 3); + + assert!(matches!( + load_scanner_cycle_state_for_startup(store).await, + ScannerCycleStateStartup::Blocked + )); + assert_eq!(scanner_cycle_recovery_status().state, "cleanup-pending"); +} + +#[test] +fn full_rescan_reset_accepts_unknown_marker_fields_without_trusting_cursor() { + let marker = br#"{ + "schema_version": 99, + "primary_revision": "memory-7", + "generation": 9000, + "leader_epoch": 9000, + "classification": "new-future-classification", + "first_detected_at_unix_secs": 1, + "last_attempt_at_unix_secs": 2, + "retry_count": 9, + "reason": "future marker", + "path": "buckets/.bloomcycle.bin", + "quarantine_path": "buckets/.bloomcycle.bin.recovery-required.json", + "future_field": {"cursor": "untrusted"} + }"#; + let decoded = + super::cycle_state::decode_recovery_marker_for_reset(marker, &DataUsageCacheRevision::Etag("memory-3".to_string())) + .expect("full-rescan compatibility decoder should accept additive fields"); + assert_eq!(decoded.primary_revision, "memory-7"); + assert_eq!(decoded.classification, "future_schema"); + assert_eq!(decoded.generation, 0); + assert_eq!(decoded.leader_epoch, 0); + assert_eq!(decoded.state, "blocked"); + + let malformed = + super::cycle_state::decode_recovery_marker_for_reset(b"{not-json", &DataUsageCacheRevision::Etag("memory-4".to_string())) + .expect("a full-rescan reset must recover even when the marker is malformed"); + assert!(malformed.primary_revision.is_empty()); + assert_eq!(malformed.classification, "future_schema"); +} + +#[tokio::test] +#[serial] +async fn full_rescan_reset_rebuilds_after_malformed_marker_without_trusting_cursor() { + let (_temp_dir, store) = setup_scanner_cycle_store().await; + save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01]) + .await + .expect("corrupt cycle state should be persisted"); + save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), br#"{not-json"#.to_vec()) + .await + .expect("malformed marker should be persisted"); + + reset_scanner_cycle_recovery(CancellationToken::new(), store.clone()) + .await + .expect("full-rescan reset should recover malformed marker"); + + let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()) + .await + .expect("rebuilt cycle state should remain durable"); + let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode"); + assert_eq!(cycle.next, 0, "reset must use the verified usage floor, not marker cursor"); + assert_eq!(leader_epoch, 1); + assert!(matches!( + read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await, + Err(EcstoreError::ConfigNotFound) + )); +} + +#[tokio::test] +#[serial] +async fn full_rescan_reset_rebuilds_when_primary_cycle_state_is_missing() { + let (_temp_dir, store) = setup_scanner_cycle_store().await; + let marker = ScannerCycleRecoveryMarker { + schema_version: 1, + primary_revision: "memory-missing".to_string(), + generation: u64::MAX, + leader_epoch: u64::MAX, + classification: "corrupt".to_string(), + first_detected_at_unix_secs: 1, + last_attempt_at_unix_secs: 2, + retry_count: 0, + reason: "missing primary".to_string(), + path: DATA_USAGE_BLOOM_NAME_PATH.clone(), + quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(), + state: "blocked".to_string(), + }; + save_config( + store.clone(), + DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), + serde_json::to_vec(&marker).expect("marker should encode"), + ) + .await + .expect("marker should be persisted"); + + reset_scanner_cycle_recovery(CancellationToken::new(), store.clone()) + .await + .expect("full-rescan reset should recreate missing primary"); + let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()) + .await + .expect("missing primary should be rebuilt"); + let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode"); + assert_eq!(cycle.next, 0); + assert_eq!(leader_epoch, 1); + assert!(matches!( + read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await, + Err(EcstoreError::ConfigNotFound) + )); +} + +#[tokio::test] +#[serial] +async fn corrupt_cycle_state_rename_or_marker_failure_stays_recovery_required() { + let store = Arc::new(MemoryConfigStore::default()); + let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()); + store.objects.lock().await.insert(state_key.clone(), vec![1]); + store.revisions.lock().await.insert(state_key, 9); + store.fail_put_number.lock().await.insert(marker_key, 1); + + assert!(matches!( + load_scanner_cycle_state_for_startup(store.clone()).await, + ScannerCycleStateStartup::Transient(_) + )); + let status = scanner_cycle_recovery_status(); + assert_eq!(status.state, "recovery-required"); + assert!(status.retryable); + assert!( + store + .objects + .lock() + .await + .contains_key(&memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str())) + ); +} + +#[tokio::test] +#[serial] +async fn oversized_or_symlinked_cycle_state_is_rejected() { + let store = Arc::new(MemoryConfigStore::default()); + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()); + store.objects.lock().await.insert(key.clone(), vec![0; 1024 * 1024 + 1]); + store.revisions.lock().await.insert(key.clone(), 11); + + assert!(matches!( + load_scanner_cycle_state_for_startup(store.clone()).await, + ScannerCycleStateStartup::Blocked + )); + assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("corrupt")); + assert!( + scanner_cycle_recovery_status() + .reason + .as_deref() + .is_some_and(|reason| reason.contains("oversized")) + ); + + let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()); + store.objects.lock().await.remove(&marker_key); + store.objects.lock().await.insert(key.clone(), vec![1]); + store.revisions.lock().await.insert(key.clone(), 12); + store.non_regular_objects.lock().await.insert(key); + // The object contract exposes a non-regular object as `is_dir`; local + // backends reject symlink/reparse entries before they become an object. + assert!(matches!( + load_scanner_cycle_state_for_startup(store).await, + ScannerCycleStateStartup::Blocked + )); +} + #[tokio::test] async fn scanner_startup_uses_primary_and_backup_usage_floor() { let store = Arc::new(MemoryConfigStore::default()); @@ -3006,6 +3333,24 @@ fn superseded_retry_backoff_grows_from_the_default_cycle() { } } +#[tokio::test(start_paused = true)] +async fn corrupt_cycle_state_backoff_uses_virtual_clock() { + let mut backoff = ScannerRetryBackoff::default(); + backoff.record_retryable_cycle(true); + let first_delay = backoff + .retry_interval(Duration::from_secs(60)) + .expect("the first recovery retry should be scheduled"); + assert_eq!(first_delay, Duration::from_secs(5)); + + let deadline = Instant::now() + first_delay; + assert!(Instant::now() < deadline); + tokio::time::advance(first_delay).await; + assert!(Instant::now() >= deadline); + + backoff.record_retryable_cycle(true); + assert_eq!(backoff.retry_interval(Duration::from_secs(60)), Some(Duration::from_secs(10))); +} + #[test] fn scanner_cycle_wait_plan_drives_growth_resets_and_bitrot_cap() { let runtime_config = ScannerRuntimeConfig { diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index f0a32f402..6822ccaa4 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -126,6 +126,7 @@ mod tests { let _list_remote_target_handler = replication::ListRemoteTargetHandler {}; let _remove_remote_target_handler = replication::RemoveRemoteTargetHandler {}; let _scanner_status_handler = scanner::ScannerStatusHandler {}; + let _scanner_cycle_state_reset_handler = scanner::ScannerCycleStateResetHandler {}; let _ilm_expiry_status_handler = scanner::IlmExpiryStatusHandler {}; let _manual_transition_handler = ilm_transition::ManualTransitionRunHandler {}; let _manual_transition_status_handler = ilm_transition::ManualTransitionJobStatusHandler {}; diff --git a/rustfs/src/admin/handlers/scanner.rs b/rustfs/src/admin/handlers/scanner.rs index ad500004d..fa8df2a69 100644 --- a/rustfs/src/admin/handlers/scanner.rs +++ b/rustfs/src/admin/handlers/scanner.rs @@ -13,8 +13,11 @@ // limitations under the License. use crate::admin::auth::authorize_admin_request; +use crate::admin::handlers::supervise_admin_mutation; use crate::admin::router::{AdminOperation, Operation, S3Router}; -use crate::admin::runtime_sources::current_scanner_metrics_report; +use crate::admin::runtime_sources::{ + app_context_from_req, current_object_store_handle_for_context, current_scanner_metrics_report, +}; use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env}; use crate::server::ADMIN_PREFIX; use chrono::Utc; @@ -22,11 +25,13 @@ use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; use matchit::Params; use rustfs_common::metrics::{ScannerLifecycleExpirySnapshot, ScannerMaintenanceControlSnapshot, ScannerMetricsReport}; +use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_credentials::Credentials; use rustfs_policy::policy::action::{Action, AdminAction}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; +use tokio_util::sync::CancellationToken; const JSON_CONTENT_TYPE: &str = "application/json"; @@ -38,6 +43,13 @@ struct ScannerStatusResponse { metrics: ScannerMetricsReport, cycle_schedule: rustfs_scanner::ScannerCycleScheduleStatus, runtime_config: rustfs_scanner::runtime_config::ScannerRuntimeConfigStatus, + cycle_recovery: rustfs_scanner::ScannerCycleRecoveryStatus, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct ScannerCycleResetRequest { + mode: String, } #[derive(Debug, Serialize)] @@ -117,6 +129,7 @@ fn scanner_status_response( metrics, cycle_schedule, runtime_config, + cycle_recovery: rustfs_scanner::scanner::scanner_cycle_recovery_status(), } } @@ -144,6 +157,11 @@ pub fn register_scanner_route(r: &mut S3Router) -> std::io::Resu format!("{ADMIN_PREFIX}/v3/scanner/status").as_str(), AdminOperation(&ScannerStatusHandler {}), )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/scanner/cycle-state/reset").as_str(), + AdminOperation(&ScannerCycleStateResetHandler {}), + )?; r.insert( Method::GET, format!("{ADMIN_PREFIX}/v3/ilm/expiry/status").as_str(), @@ -163,6 +181,13 @@ async fn validate_scanner_status_request(req: &S3Request) -> S3Result) -> S3Result { + if req.credentials.is_none() { + return Err(s3_error!(InvalidRequest, "missing credentials")); + } + authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ConfigUpdateAdminAction)]).await +} + fn json_response(body: Vec) -> S3Result> { let mut headers = HeaderMap::new(); let content_type = HeaderValue::from_str(JSON_CONTENT_TYPE) @@ -192,6 +217,37 @@ impl Operation for ScannerStatusHandler { pub struct IlmExpiryStatusHandler {} +pub struct ScannerCycleStateResetHandler {} + +#[async_trait::async_trait] +impl Operation for ScannerCycleStateResetHandler { + async fn call(&self, mut req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let _cred = validate_scanner_reset_request(&req).await?; + let body = req + .input + .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE) + .await + .map_err(|err| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid reset request body: {err}")))?; + let reset = serde_json::from_slice::(&body) + .map_err(|err| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid reset request body: {err}")))?; + if reset.mode != "full-rescan" { + return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "reset mode must be full-rescan")); + } + let context = app_context_from_req(&req) + .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "storage layer not initialized"))?; + let store = current_object_store_handle_for_context(Some(context.as_ref())) + .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "storage layer not initialized"))?; + supervise_admin_mutation("scanner cycle state reset", async move { + rustfs_scanner::scanner::reset_scanner_cycle_recovery(CancellationToken::new(), store) + .await + .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))?; + Ok::<_, S3Error>(()) + }) + .await?; + json_response(br#"{"status":"reset","mode":"full-rescan"}"#.to_vec()) + } +} + #[async_trait::async_trait] impl Operation for IlmExpiryStatusHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -237,6 +293,38 @@ mod tests { assert_eq!(err.message(), Some("missing credentials")); } + #[tokio::test] + async fn scanner_reset_gate_rejects_missing_credentials() { + let req = S3Request { + input: Body::from(String::new()), + method: Method::POST, + uri: http::Uri::from_static("/rustfs/admin/v3/scanner/cycle-state/reset"), + headers: HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + + let err = validate_scanner_reset_request(&req) + .await + .expect_err("a reset request without credentials must be rejected"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + assert_eq!(err.message(), Some("missing credentials")); + } + + #[test] + fn admin_reset_requires_full_rescan_or_verified_cursor() { + let full_rescan: ScannerCycleResetRequest = + serde_json::from_str(r#"{"mode":"full-rescan"}"#).expect("full rescan must be accepted"); + assert_eq!(full_rescan.mode, "full-rescan"); + let cursor: ScannerCycleResetRequest = + serde_json::from_str(r#"{"mode":"cursor"}"#).expect("mode validation belongs to the handler"); + assert_ne!(cursor.mode, "full-rescan"); + assert!(serde_json::from_str::(r#"{"mode":"full-rescan","cursor":"untrusted"}"#).is_err()); + } + #[test] fn scanner_disabled_reason_reports_startup_env_key() { assert_eq!(scanner_disabled_reason(true), None); @@ -304,6 +392,11 @@ mod tests { assert_eq!(encoded["cycle_schedule"]["effective_interval_seconds"], 0); assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_enabled"], false); assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_multiplier"], 1); + assert_eq!(encoded["cycle_recovery"]["state"], "healthy"); + assert_eq!( + encoded["cycle_recovery"]["quarantine_path"], + rustfs_scanner::DATA_USAGE_BLOOM_RECOVERY_PATH.as_str() + ); } #[test] diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index ccb013211..e423c0b0e 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -428,6 +428,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ admin(HttpMethod::Get, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High), admin(HttpMethod::Put, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High), admin(HttpMethod::Get, "/rustfs/admin/v3/scanner/status", SERVER_INFO, RouteRiskLevel::Sensitive), + admin( + HttpMethod::Post, + "/rustfs/admin/v3/scanner/cycle-state/reset", + CONFIG_UPDATE, + RouteRiskLevel::High, + ), admin( HttpMethod::Get, "/rustfs/admin/v3/ilm/expiry/status", @@ -2020,6 +2026,12 @@ mod tests { assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/expiry/status", SET_TIER); } + #[test] + fn route_policy_requires_config_update_for_scanner_cycle_reset() { + assert_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", CONFIG_UPDATE); + assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", SERVER_INFO); + } + #[test] fn route_policy_uses_tier_actions_for_transition_routes() { assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER); diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index e48e09c94..8dcb5d901 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -243,6 +243,7 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::GET, "/v3/config"), admin_route(Method::PUT, "/v3/config"), admin_route(Method::GET, "/v3/scanner/status"), + admin_route(Method::POST, "/v3/scanner/cycle-state/reset"), admin_route(Method::GET, "/v3/audit/target/list"), admin_route_sample( Method::PUT, @@ -879,6 +880,7 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::GET, &admin_path("/v3/config")); assert_route(&router, Method::PUT, &admin_path("/v3/config")); assert_route(&router, Method::GET, &admin_path("/v3/scanner/status")); + assert_route(&router, Method::POST, &admin_path("/v3/scanner/cycle-state/reset")); assert_route(&router, Method::GET, &admin_path("/v3/ilm/expiry/status")); assert_route(&router, Method::POST, &admin_path("/v3/ilm/transition/run")); assert_route( @@ -1367,6 +1369,7 @@ fn test_admin_alias_paths_match_existing_admin_routes() { (Method::GET, compat_admin_alias_path("/v3/config")), (Method::PUT, compat_admin_alias_path("/v3/config")), (Method::GET, compat_admin_alias_path("/v3/scanner/status")), + (Method::POST, compat_admin_alias_path("/v3/scanner/cycle-state/reset")), (Method::GET, compat_admin_alias_path("/v3/ilm/expiry/status")), ] { assert!(