From ab44ae7e83d3f44a9a8fb9292f5746f03a959e0f Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 1 Sep 2026 09:51:46 +0800 Subject: [PATCH] fix(scanner): add supported usage state reset (#6972) Add an authenticated scanner usage-state reset endpoint that publishes a fenced bootstrap marker for full rebuilds instead of requiring operators to delete usage metadata by hand. Guard the reset with the scanner leader lock, storage publication epoch, and per-slot revision preconditions, and make startup resumable across stale cleanup leftovers while still rejecting newer conflicting usage state. Co-authored-by: heihutu Co-authored-by: Zhengchao An --- crates/scanner/src/lib.rs | 6 +- crates/scanner/src/scanner.rs | 3 +- crates/scanner/src/scanner/cycle_state.rs | 605 ++++++++++++++++---- crates/scanner/src/scanner/tests.rs | 259 +++++++++ docs/operations/scanner-runtime-controls.md | 51 ++ rustfs/src/admin/handlers/mod.rs | 1 + rustfs/src/admin/handlers/scanner.rs | 61 ++ rustfs/src/admin/route_policy.rs | 12 + rustfs/src/admin/route_registration_test.rs | 3 + 9 files changed, 898 insertions(+), 103 deletions(-) diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index e2992e95a..a1200455c 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -83,9 +83,9 @@ pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_st pub use rustfs_scanner_contracts::last_minute; pub use scanner::{ ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, - ScannerPauseBacklogPhase, ScannerPauseBacklogStatus, ScannerPauseBacklogThresholds, init_data_scanner, - reset_scanner_cycle_recovery, scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_pause_backlog_status, - scanner_topology_digest, + 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, }; pub use scanner_io::{ ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index b79965d90..d687c1081 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -3385,7 +3385,8 @@ pub use backlog::{ #[cfg(test)] pub(crate) use cycle_state::encode_scanner_cycle_fence_for_test; pub use cycle_state::{ - ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, reset_scanner_cycle_recovery, scanner_cycle_recovery_status, + ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerUsageStateResetResult, reset_scanner_cycle_recovery, + reset_scanner_usage_state_for_full_rebuild, scanner_cycle_recovery_status, }; pub(crate) use cycle_state::{ current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence, load_scanner_cycle_state_for_startup, diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 99d2958d5..5788034ce 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -29,6 +29,8 @@ const USAGE_FLOOR_LOAD_FAILED: &str = "usage_floor_load_failed"; const LEGACY_EMPTY_USAGE_FLOOR_RECOVERY: &str = "legacy_empty_usage_floor"; const CACHE_CYCLE_AHEAD: &str = "cache_cycle_ahead"; +const SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD: &str = "full-rebuild"; + #[derive(Clone, Debug, Default, Serialize)] pub struct ScannerCycleRecoveryStatus { /// The immutable primary object whose revision is being guarded. @@ -50,6 +52,23 @@ pub struct ScannerCycleRecoveryStatus { pub reason: Option, } +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +pub struct ScannerUsageStateResetResult { + pub status: String, + pub mode: String, + pub usage_state: String, + pub leader_epoch: u64, + pub next_cycle: u64, + pub reset_paths: Vec, +} + +#[derive(Clone, Debug)] +pub(super) struct ScannerUsageStateResetSlot { + path: String, + data: Option>, + revision: DataUsageCacheRevision, +} + static SCANNER_CYCLE_RECOVERY_STATUS: LazyLock> = LazyLock::new(|| { RwLock::new(ScannerCycleRecoveryStatus { path: DATA_USAGE_BLOOM_NAME_PATH.clone(), @@ -1288,6 +1307,292 @@ pub async fn reset_scanner_cycle_recovery(ctx: CancellationToken, storeapi: Arc< Ok(()) } +fn scanner_usage_state_reset_paths() -> Vec { + vec![ + DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(), + format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()), + LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(), + format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str().to_string(), + ] +} + +pub(super) async fn read_usage_state_reset_slots( + storeapi: Arc, +) -> Result, ScannerError> { + let mut slots = Vec::new(); + for path in scanner_usage_state_reset_paths() { + let (data, revision) = read_config_with_revision(storeapi.clone(), &path) + .await + .map_err(|err| ScannerError::Other(format!("failed to inspect scanner usage reset slot {path}: {err}")))?; + slots.push(ScannerUsageStateResetSlot { path, data, revision }); + } + Ok(slots) +} + +fn usage_state_reset_floor(slots: &[ScannerUsageStateResetSlot]) -> Result { + let mut floor = PersistedUsageFloor::default(); + for slot in slots { + let Some(data) = slot.data.as_deref() else { + continue; + }; + let Ok(usage) = serde_json::from_slice::(data) else { + continue; + }; + update_persisted_usage_floor(&mut floor, &usage, &slot.path)?; + } + Ok(floor) +} + +async fn read_cycle_state_for_usage_reset( + storeapi: Arc, +) -> Result<(CurrentCycle, u64, DataUsageCacheRevision), ScannerError> { + 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, + ) => return Ok((CurrentCycle::default(), 0, DataUsageCacheRevision::Missing)), + Err(err) => return Err(ScannerError::Other(format!("failed to inspect scanner cycle state: {err}"))), + }; + if reader.object_info.is_dir || reader.object_info.size < 0 { + return Err(ScannerError::Other( + "scanner usage reset requires a regular scanner cycle state object".to_string(), + )); + } + let revision = reader + .object_info + .etag + .as_ref() + .filter(|etag| !etag.is_empty()) + .cloned() + .map(DataUsageCacheRevision::Etag) + .ok_or_else(|| ScannerError::Other("scanner cycle state has no revision".to_string()))?; + let data = read_cycle_state_body(&mut reader) + .await + .map_err(|err| ScannerError::Other(format!("failed to read scanner cycle state for usage reset: {err}")))?; + let (cycle, leader_epoch) = decode_scanner_cycle_state(&data).map_err(|err| { + ScannerError::Other(format!( + "scanner usage reset requires a valid scanner cycle state; reset scanner cycle state first: {err}" + )) + })?; + Ok((cycle, leader_epoch, revision)) +} + +async fn delete_usage_state_reset_slot( + storeapi: Arc, + slot: &ScannerUsageStateResetSlot, + expected_epoch: u64, +) -> Result { + if matches!(slot.revision, DataUsageCacheRevision::Missing) { + return Ok(false); + } + let delete_result = delete_config_with_publication_admission_for_epoch( + storeapi.clone(), + RUSTFS_META_BUCKET, + &slot.path, + ScannerObjectOptions { + delete_prefix: false, + http_preconditions: Some(slot.revision.preconditions()), + ..Default::default() + }, + expected_epoch, + ) + .await; + match delete_result { + Ok(_) => Ok(true), + Err(EcstoreError::FileNotFound | EcstoreError::ConfigNotFound | EcstoreError::ObjectNotFound(_, _)) => Ok(false), + Err(EcstoreError::PreconditionFailed) => Err(ScannerError::Other(format!( + "scanner usage reset slot changed while deleting {}", + slot.path + ))), + Err(err) if scanner_publication_epoch_changed(&err) => { + Err(ScannerError::Other("scanner usage reset deferred by a movement epoch change".to_string())) + } + Err(err) => Err(ScannerError::Other(format!( + "failed to delete scanner usage reset slot {}: {err}", + slot.path + ))), + } +} + +pub(super) async fn reset_scanner_usage_state_slots_for_full_rebuild( + storeapi: Arc, + slots: &[ScannerUsageStateResetSlot], + expected_epoch: u64, + leader_epoch: u64, +) -> Result, ScannerError> { + let mut reset_paths = Vec::new(); + let primary = slots + .iter() + .find(|slot| slot.path == DATA_USAGE_OBJ_NAME_PATH.as_str()) + .ok_or_else(|| ScannerError::Other("scanner usage reset primary slot was not inspected".to_string()))?; + let marker = DataUsageInfo { + last_update: Some(std::time::SystemTime::now()), + scanner_epoch: Some(leader_epoch), + usage_snapshot_converged: Some(false), + usage_snapshot_bootstrap_pending: true, + ..Default::default() + }; + let data = serde_json::to_vec(&marker) + .map_err(|err| ScannerError::Other(format!("failed to encode scanner usage reset bootstrap marker: {err}")))?; + let save_result = save_config_with_publication_admission_for_epoch( + storeapi.clone(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + data.clone(), + primary.revision.preconditions(), + expected_epoch, + ) + .await; + if save_result + .as_ref() + .ok() + .and_then(|info| info.etag.as_deref()) + .is_some_and(|etag| !etag.is_empty()) + { + reset_paths.push(DATA_USAGE_OBJ_NAME_PATH.as_str().to_string()); + } else { + let (persisted, revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .map_err(|err| ScannerError::Other(format!("failed to reconcile scanner usage reset bootstrap marker: {err}")))?; + if persisted.as_deref() != Some(data.as_slice()) || !matches!(revision, DataUsageCacheRevision::Etag(_)) { + return Err(ScannerError::Other(match save_result { + Ok(_) => "scanner usage reset bootstrap returned no ETag and could not be confirmed".to_string(), + Err(err) if scanner_publication_epoch_changed(&err) => { + "scanner usage reset deferred by a movement epoch change".to_string() + } + Err(EcstoreError::PreconditionFailed) => { + "scanner usage reset primary slot changed before bootstrap publish".to_string() + } + Err(err) => format!("failed to persist scanner usage reset bootstrap: {err}"), + })); + } + reset_paths.push(DATA_USAGE_OBJ_NAME_PATH.as_str().to_string()); + } + + for slot in slots.iter().filter(|slot| slot.path != DATA_USAGE_OBJ_NAME_PATH.as_str()) { + if delete_usage_state_reset_slot(storeapi.clone(), slot, expected_epoch).await? { + reset_paths.push(slot.path.clone()); + } + } + invalidate_admin_data_usage_snapshot_cache().await; + invalidate_data_usage_snapshot_cache().await; + Ok(reset_paths) +} + +pub async fn reset_scanner_usage_state_for_full_rebuild( + ctx: CancellationToken, + storeapi: Arc, +) -> Result { + 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 usage reset".to_string())); + } + if ctx.is_cancelled() { + return Err(ScannerError::Other("scanner usage reset was cancelled".to_string())); + } + + let Some(reset_epoch) = scanner_publication_epoch(storeapi.clone()).await else { + return Err(ScannerError::Other("scanner usage reset is blocked by data movement".to_string())); + }; + let (cycle, cycle_epoch, cycle_revision) = read_cycle_state_for_usage_reset(storeapi.clone()).await?; + let slots = read_usage_state_reset_slots(storeapi.clone()).await?; + let usage_floor = usage_state_reset_floor(&slots)?; + let leader_epoch = cycle_epoch + .max(usage_floor.leader_epoch) + .checked_add(1) + .filter(|epoch| *epoch < u64::MAX) + .ok_or_else(|| ScannerError::Other("scanner leader epoch is exhausted".to_string()))?; + let rebuilt_cycle = CurrentCycle { + next: cycle.next.max(usage_floor.next_cycle), + ..Default::default() + }; + let cycle_data = encode_scanner_cycle_state(&rebuilt_cycle, leader_epoch) + .map_err(|err| ScannerError::Other(format!("failed to encode scanner cycle state for usage reset: {err}")))?; + + if guard.is_lock_lost() { + return Err(ScannerError::Other( + "scanner leader lock was lost before fencing usage reset cycle state".to_string(), + )); + } + save_config_with_publication_admission_for_epoch( + storeapi.clone(), + DATA_USAGE_BLOOM_NAME_PATH.as_str(), + cycle_data, + cycle_revision.preconditions(), + reset_epoch, + ) + .await + .map_err(|err| { + if scanner_publication_epoch_changed(&err) { + ScannerError::Other("scanner usage reset deferred by a movement epoch change".to_string()) + } else { + ScannerError::Other(format!("failed to fence scanner cycle state for usage reset: {err}")) + } + })?; + + if guard.is_lock_lost() { + return Err(ScannerError::Other( + "scanner leader lock was lost before publishing usage reset marker".to_string(), + )); + } + let reset_paths = + reset_scanner_usage_state_slots_for_full_rebuild(storeapi.clone(), &slots, reset_epoch, leader_epoch).await?; + if guard.is_lock_lost() { + return Err(ScannerError::Other( + "scanner leader lock was lost after publishing usage reset marker".to_string(), + )); + } + + clear_scanner_usage_floor_failure(); + clear_legacy_empty_usage_floor_recovery_status(); + set_scanner_cycle_recovery_status(recovery_status("healthy", None, false)); + super::notify_scanner_cycle_recovery_wake(); + info!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "usage_state_reset", + mode = SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD, + leader_epoch, + next_cycle = rebuilt_cycle.next, + reset_paths = reset_paths.len(), + "Scanner usage state reset was published" + ); + + Ok(ScannerUsageStateResetResult { + status: "reset".to_string(), + mode: SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD.to_string(), + usage_state: "bootstrap-pending".to_string(), + leader_epoch, + next_cycle: rebuilt_cycle.next, + reset_paths, + }) +} + #[derive(Debug, thiserror::Error)] pub(super) enum ScannerCycleStateError { #[error("failed to encode scanner cycle state: {0}")] @@ -1681,6 +1986,97 @@ pub(super) async fn persisted_usage_floor( Ok(floor) } +fn usage_epoch(usage: &DataUsageInfo) -> u64 { + usage.scanner_epoch.unwrap_or_default() +} + +fn usage_is_older_than_bootstrap(usage: &DataUsageInfo, bootstrap_epoch: Option) -> bool { + bootstrap_epoch.is_some_and(|epoch| usage_epoch(usage) < epoch) +} + +fn decode_usage_floor_slot(data: &[u8], path: &str) -> Result { + serde_json::from_slice::(data) + .map_err(|err| ScannerError::Other(format!("failed to decode scanner usage floor from {path}: {err}"))) +} + +fn update_persisted_usage_floor(floor: &mut PersistedUsageFloor, usage: &DataUsageInfo, path: &str) -> Result<(), ScannerError> { + floor.leader_epoch = floor.leader_epoch.max(usage_epoch(usage)); + if let Some(completed_cycle) = usage.scanner_cycle { + let next_cycle = completed_cycle + .checked_add(1) + .filter(|next| *next < u64::MAX) + .ok_or_else(|| ScannerError::Other(format!("persisted scanner usage cycle is exhausted in {path}")))?; + floor.next_cycle = floor.next_cycle.max(next_cycle); + } + Ok(()) +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum BootstrapBackupAction { + Resume, + StopV2Pair, +} + +struct BootstrapBackupSlot<'a> { + data: &'a [u8], + usage: &'a DataUsageInfo, + backup_path: &'a str, + primary_path: &'a str, + bootstrap_epoch: Option, + recovered_bootstrap: bool, + recovered_primary_companion_epoch: Option, +} + +struct BootstrapBackupResolution<'a> { + floor: &'a mut PersistedUsageFloor, + unrecoverable_baseline_path: &'a mut Option, +} + +fn resolve_bootstrap_backup_slot( + slot: BootstrapBackupSlot<'_>, + resolution: BootstrapBackupResolution<'_>, +) -> Result { + if usage_is_older_than_bootstrap(slot.usage, slot.bootstrap_epoch) { + return Ok(BootstrapBackupAction::Resume); + } + if let Some(primary_epoch) = slot.recovered_primary_companion_epoch { + if data_usage_info_has_persisted_baseline_identity(slot.usage) { + let backup_epoch = usage_epoch(slot.usage); + if backup_epoch >= primary_epoch { + update_persisted_usage_floor(resolution.floor, slot.usage, slot.backup_path)?; + } + } else if let Some(epoch) = legacy_empty_usage_fence_epoch(slot.data, slot.usage) { + if let Some(epoch) = epoch { + resolution.floor.leader_epoch = resolution.floor.leader_epoch.max(epoch); + } + } else { + resolution + .unrecoverable_baseline_path + .get_or_insert_with(|| slot.backup_path.to_string()); + } + return Ok(BootstrapBackupAction::Resume); + } + let compatible_empty_fence = legacy_empty_usage_fence_epoch(slot.data, slot.usage).is_some_and(|epoch| { + epoch.is_none_or(|epoch| slot.bootstrap_epoch.is_some_and(|bootstrap_epoch| epoch <= bootstrap_epoch)) + }); + if compatible_empty_fence { + return Ok(BootstrapBackupAction::Resume); + } + if slot.recovered_bootstrap && data_usage_info_has_persisted_baseline_identity(slot.usage) { + let epoch = usage_epoch(slot.usage); + if epoch >= resolution.floor.leader_epoch { + update_persisted_usage_floor(resolution.floor, slot.usage, slot.backup_path)?; + if slot.primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str() { + return Ok(BootstrapBackupAction::StopV2Pair); + } + return Ok(BootstrapBackupAction::Resume); + } + } + Err(ScannerError::Other( + "scanner usage bootstrap conflicts with a persisted backup".to_string(), + )) +} + pub(super) async fn persisted_usage_floor_for_startup( storeapi: Arc, allow_missing_for_bootstrap: bool, @@ -1704,17 +2100,6 @@ pub(super) async fn persisted_usage_floor_for_startup( let mut unrecoverable_baseline_path: Option = None; let mut stale_authoritative_path: Option = None; let mut legacy_empty_primary: Option = None; - let update_floor = |floor: &mut PersistedUsageFloor, usage: &DataUsageInfo, path: &str| -> Result<(), ScannerError> { - floor.leader_epoch = floor.leader_epoch.max(usage.scanner_epoch.unwrap_or_default()); - if let Some(completed_cycle) = usage.scanner_cycle { - let next_cycle = completed_cycle - .checked_add(1) - .filter(|next| *next < u64::MAX) - .ok_or_else(|| ScannerError::Other(format!("persisted scanner usage cycle is exhausted in {path}")))?; - floor.next_cycle = floor.next_cycle.max(next_cycle); - } - Ok(()) - }; for primary_path in [DATA_USAGE_OBJ_NAME_PATH.as_str(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()] { let backup_path = format!("{primary_path}.bkp"); let is_v2_path = primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str(); @@ -1722,60 +2107,82 @@ pub(super) async fn persisted_usage_floor_for_startup( let mut primary_read_error = None; let primary_epoch = match read_config_with_revision(storeapi.clone(), primary_path).await { Ok((Some(data), revision)) => { - let usage = serde_json::from_slice::(&data).map_err(|err| { - ScannerError::Other(format!("failed to decode scanner usage floor from {primary_path}: {err}")) - })?; - if data_usage_info_is_bootstrap_pending(&usage) && primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str() { - if bootstrap_pending { - return Err(ScannerError::Other("multiple scanner usage bootstrap markers were found".to_string())); - } - bootstrap_pending = true; - bootstrap_epoch = usage.scanner_epoch; - if let Some((marker, _)) = recovery_marker.as_ref() { - if usage.scanner_epoch.is_none_or(|epoch| epoch < marker.leader_epoch) { - return Err(ScannerError::Other( - "scanner usage bootstrap is older than its recovery marker".to_string(), - )); - } - recovered_bootstrap = true; - } - update_floor(&mut floor, &usage, primary_path)?; - None - } else if !data_usage_info_has_persisted_baseline_identity(&usage) { - invalid_baseline_path.get_or_insert_with(|| primary_path.to_string()); - invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch); - match legacy_empty_usage_fence_epoch(&data, &usage) { - Some(Some(epoch)) if is_v2_path => { - legacy_empty_primary = Some(LegacyEmptyUsageFloorPrimary { revision, epoch }); - } - Some(_) => {} - None => { - unrecoverable_baseline_path.get_or_insert_with(|| primary_path.to_string()); - } - } - None - } else { - let epoch = usage.scanner_epoch.unwrap_or_default(); - if recovered_bootstrap && !is_v2_path { - if epoch < floor.leader_epoch { - unrecoverable_baseline_path.get_or_insert_with(|| primary_path.to_string()); - } else { - update_floor(&mut floor, &usage, primary_path)?; - recovered_primary_companion_epoch = Some(epoch); - } + let usage = match decode_usage_floor_slot(&data, primary_path) { + Ok(usage) => Some(usage), + Err(err) if bootstrap_pending && !is_v2_path => { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "usage_reset_cleanup_deferred", + path = %primary_path, + error = %err, + "Scanner usage reset ignored a stale legacy usage primary after bootstrap marker" + ); None - // A legacy snapshot may be structurally valid but older - // than an incomplete v2 snapshot left by a newer leader. - // Do not let that candidate regress the startup floor. - } else if invalid_baseline_epoch.is_some_and(|fenced_epoch| epoch < fenced_epoch) - && (!is_v2_path || recovery_marker.is_some()) - { - stale_authoritative_path.get_or_insert_with(|| primary_path.to_string()); + } + Err(err) => { + return Err(err); + } + }; + if let Some(usage) = usage { + if data_usage_info_is_bootstrap_pending(&usage) && primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str() { + if bootstrap_pending { + return Err(ScannerError::Other("multiple scanner usage bootstrap markers were found".to_string())); + } + bootstrap_pending = true; + bootstrap_epoch = usage.scanner_epoch; + if let Some((marker, _)) = recovery_marker.as_ref() { + if usage.scanner_epoch.is_none_or(|epoch| epoch < marker.leader_epoch) { + return Err(ScannerError::Other( + "scanner usage bootstrap is older than its recovery marker".to_string(), + )); + } + recovered_bootstrap = true; + } + update_persisted_usage_floor(&mut floor, &usage, primary_path)?; + None + } else if bootstrap_pending && !is_v2_path && usage_is_older_than_bootstrap(&usage, bootstrap_epoch) { + None + } else if !data_usage_info_has_persisted_baseline_identity(&usage) { + invalid_baseline_path.get_or_insert_with(|| primary_path.to_string()); + invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch); + match legacy_empty_usage_fence_epoch(&data, &usage) { + Some(Some(epoch)) if is_v2_path => { + legacy_empty_primary = Some(LegacyEmptyUsageFloorPrimary { revision, epoch }); + } + Some(_) => {} + None => { + unrecoverable_baseline_path.get_or_insert_with(|| primary_path.to_string()); + } + } None } else { - update_floor(&mut floor, &usage, primary_path)?; - Some(epoch) + let epoch = usage_epoch(&usage); + if recovered_bootstrap && !is_v2_path { + if epoch < floor.leader_epoch { + unrecoverable_baseline_path.get_or_insert_with(|| primary_path.to_string()); + } else { + update_persisted_usage_floor(&mut floor, &usage, primary_path)?; + recovered_primary_companion_epoch = Some(epoch); + } + None + // A legacy snapshot may be structurally valid but older + // than an incomplete v2 snapshot left by a newer leader. + // Do not let that candidate regress the startup floor. + } else if invalid_baseline_epoch.is_some_and(|fenced_epoch| epoch < fenced_epoch) + && (!is_v2_path || recovery_marker.is_some()) + { + stale_authoritative_path.get_or_insert_with(|| primary_path.to_string()); + None + } else { + update_persisted_usage_floor(&mut floor, &usage, primary_path)?; + Some(epoch) + } } + } else { + None } } Ok((None, _)) => None, @@ -1794,44 +2201,44 @@ pub(super) async fn persisted_usage_floor_for_startup( let mut any_found = primary_epoch.is_some(); match read_config_with_revision(storeapi.clone(), &backup_path).await { Ok((Some(data), _)) => { - let usage = serde_json::from_slice::(&data).map_err(|err| { - ScannerError::Other(format!("failed to decode scanner usage floor from {backup_path}: {err}")) - })?; + let usage = match decode_usage_floor_slot(&data, &backup_path) { + Ok(usage) => usage, + Err(err) if bootstrap_pending => { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "usage_reset_cleanup_deferred", + path = %backup_path, + error = %err, + "Scanner usage reset ignored a stale usage backup after bootstrap marker" + ); + continue; + } + Err(err) => { + return Err(err); + } + }; if bootstrap_pending { - if let Some(primary_epoch) = recovered_primary_companion_epoch { - if data_usage_info_has_persisted_baseline_identity(&usage) { - let backup_epoch = usage.scanner_epoch.unwrap_or_default(); - if backup_epoch >= primary_epoch { - update_floor(&mut floor, &usage, &backup_path)?; - } - } else if let Some(epoch) = legacy_empty_usage_fence_epoch(&data, &usage) { - if let Some(epoch) = epoch { - floor.leader_epoch = floor.leader_epoch.max(epoch); - } - } else { - unrecoverable_baseline_path.get_or_insert_with(|| backup_path.clone()); - } - continue; + match resolve_bootstrap_backup_slot( + BootstrapBackupSlot { + data: &data, + usage: &usage, + backup_path: &backup_path, + primary_path, + bootstrap_epoch, + recovered_bootstrap, + recovered_primary_companion_epoch, + }, + BootstrapBackupResolution { + floor: &mut floor, + unrecoverable_baseline_path: &mut unrecoverable_baseline_path, + }, + )? { + BootstrapBackupAction::Resume => continue, + BootstrapBackupAction::StopV2Pair => break, } - let compatible_empty_fence = legacy_empty_usage_fence_epoch(&data, &usage).is_some_and(|epoch| { - epoch.is_none_or(|epoch| bootstrap_epoch.is_some_and(|bootstrap_epoch| epoch <= bootstrap_epoch)) - }); - if compatible_empty_fence { - continue; - } - if recovered_bootstrap && data_usage_info_has_persisted_baseline_identity(&usage) { - let epoch = usage.scanner_epoch.unwrap_or_default(); - if epoch >= floor.leader_epoch { - update_floor(&mut floor, &usage, &backup_path)?; - if primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str() { - break; - } - continue; - } - } - return Err(ScannerError::Other( - "scanner usage bootstrap conflicts with a persisted backup".to_string(), - )); } if !data_usage_info_has_persisted_baseline_identity(&usage) { invalid_baseline_path.get_or_insert_with(|| backup_path.clone()); @@ -1843,14 +2250,14 @@ pub(super) async fn persisted_usage_floor_for_startup( // missing-state bootstrap. Continue to a legacy pair in // case it contains a complete, fenced snapshot. } else { - let backup_epoch = usage.scanner_epoch.unwrap_or_default(); + let backup_epoch = usage_epoch(&usage); // A backup write from an older leader may complete after the // primary epoch has been fenced. It must not advance the startup // floor unless its epoch is at least as new as the primary. if primary_epoch.is_none_or(|epoch| backup_epoch >= epoch) && invalid_baseline_epoch.is_none_or(|epoch| backup_epoch >= epoch) { - update_floor(&mut floor, &usage, &backup_path)?; + update_persisted_usage_floor(&mut floor, &usage, &backup_path)?; any_found = true; } else { stale_authoritative_path.get_or_insert_with(|| backup_path.clone()); diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 29a65a0e0..263501089 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -4377,6 +4377,265 @@ async fn usage_bootstrap_does_not_overwrite_concurrent_replacement() { ); } +#[tokio::test] +#[serial] +async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() { + let (_temp_dir, store) = setup_scanner_cycle_store().await; + let cycle = CurrentCycle { + current: 41, + next: 42, + cycle_completed: vec![Utc::now()], + started: Utc::now(), + }; + save_config( + store.clone(), + DATA_USAGE_BLOOM_NAME_PATH.as_str(), + encode_scanner_cycle_state(&cycle, 7).expect("cycle state should encode"), + ) + .await + .expect("cycle state should persist"); + + let usage_backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); + let legacy_backup_path = format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()); + for (path, epoch, cycle) in [ + (usage_backup_path.as_str(), 6, 40), + (LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), 5, 39), + (legacy_backup_path.as_str(), 4, 38), + (DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), 8, 41), + ] { + let mut usage = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + usage.scanner_epoch = Some(epoch); + usage.scanner_cycle = Some(cycle); + save_config(store.clone(), path, serde_json::to_vec(&usage).expect("usage slot should encode")) + .await + .expect("usage slot should persist"); + } + + let result = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store.clone()) + .await + .expect("usage state reset should publish a fenced bootstrap marker"); + + assert_eq!(result.status, "reset"); + assert_eq!(result.mode, "full-rebuild"); + assert_eq!(result.usage_state, "bootstrap-pending"); + assert_eq!(result.leader_epoch, 9); + assert_eq!(result.next_cycle, 42); + assert_eq!(result.reset_paths.len(), 5); + + let cycle_state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()) + .await + .expect("reset cycle state should remain"); + let (reset_cycle, reset_epoch) = decode_scanner_cycle_state(&cycle_state).expect("reset cycle state should decode"); + assert_eq!(reset_cycle.next, 42); + assert_eq!(reset_cycle.current, 0); + assert!(reset_cycle.cycle_completed.is_empty()); + assert_eq!(reset_epoch, 9); + + let usage = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .expect("reset usage marker should remain"); + let usage = serde_json::from_slice::(&usage).expect("reset usage marker should decode"); + assert!(data_usage_info_is_bootstrap_pending(&usage)); + assert!(!data_usage_info_has_persisted_baseline_identity(&usage)); + assert_eq!(usage.scanner_epoch, Some(9)); + + for path in [ + usage_backup_path.as_str(), + LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), + legacy_backup_path.as_str(), + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + ] { + assert!( + matches!(read_config(store.clone(), path).await, Err(EcstoreError::ConfigNotFound)), + "reset should remove stale usage slot {path}" + ); + } + + let (floor, state) = persisted_usage_floor_for_startup(store, false) + .await + .expect("reset marker should be resumable"); + assert_eq!(floor.leader_epoch, 9); + assert_eq!(state, PersistedUsageFloorStartup::BootstrapPending); +} + +#[tokio::test] +async fn scanner_usage_state_reset_bootstrap_survives_stale_cleanup_slots_after_restart() { + let store = Arc::new(MemoryConfigStore::default()); + let mut marker = DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH), + scanner_epoch: Some(9), + usage_snapshot_converged: Some(false), + usage_snapshot_bootstrap_pending: true, + ..Default::default() + }; + save_config( + store.clone(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + serde_json::to_vec(&marker).expect("usage reset marker should encode"), + ) + .await + .expect("usage reset marker should persist"); + save_config( + store.clone(), + format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()).as_str(), + b"{not-json".to_vec(), + ) + .await + .expect("stale malformed backup should persist"); + + marker.usage_snapshot_bootstrap_pending = false; + marker.usage_snapshot_complete = true; + marker.scanner_epoch = Some(8); + marker.scanner_cycle = Some(41); + for path in [ + LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(), + format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + ] { + save_config( + store.clone(), + &path, + serde_json::to_vec(&marker).expect("stale legacy usage should encode"), + ) + .await + .expect("stale legacy usage should persist"); + } + + let (floor, state) = persisted_usage_floor_for_startup(store.clone(), false) + .await + .expect("restart should resume reset bootstrap while stale cleanup slots remain"); + assert_eq!( + floor, + PersistedUsageFloor { + next_cycle: 0, + leader_epoch: 9, + } + ); + assert_eq!(state, PersistedUsageFloorStartup::BootstrapPending); + assert!( + persisted_usage_floor(store).await.is_err(), + "bootstrap marker must still not become an authoritative usage floor" + ); +} + +#[tokio::test] +async fn scanner_usage_state_reset_bootstrap_survives_malformed_legacy_primary_after_restart() { + let store = Arc::new(MemoryConfigStore::default()); + let marker = DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH), + scanner_epoch: Some(9), + usage_snapshot_converged: Some(false), + usage_snapshot_bootstrap_pending: true, + ..Default::default() + }; + save_config( + store.clone(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + serde_json::to_vec(&marker).expect("usage reset marker should encode"), + ) + .await + .expect("usage reset marker should persist"); + save_config(store.clone(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), b"{not-json".to_vec()) + .await + .expect("stale malformed legacy primary should persist"); + + let (floor, state) = persisted_usage_floor_for_startup(store.clone(), false) + .await + .expect("restart should resume reset bootstrap when only stale malformed legacy primary remains"); + assert_eq!( + floor, + PersistedUsageFloor { + next_cycle: 0, + leader_epoch: 9, + } + ); + assert_eq!(state, PersistedUsageFloorStartup::BootstrapPending); + assert!(persisted_usage_floor(store).await.is_err()); +} + +#[tokio::test] +async fn scanner_usage_state_reset_bootstrap_does_not_mask_newer_legacy_backup() { + let store = Arc::new(MemoryConfigStore::default()); + let marker = DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH), + scanner_epoch: Some(9), + usage_snapshot_converged: Some(false), + usage_snapshot_bootstrap_pending: true, + ..Default::default() + }; + save_config( + store.clone(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + serde_json::to_vec(&marker).expect("usage reset marker should encode"), + ) + .await + .expect("usage reset marker should persist"); + save_config(store.clone(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), b"{not-json".to_vec()) + .await + .expect("malformed legacy primary should persist"); + + let mut newer_backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + newer_backup.scanner_epoch = Some(10); + newer_backup.scanner_cycle = Some(43); + save_config( + store.clone(), + format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()).as_str(), + serde_json::to_vec(&newer_backup).expect("newer legacy backup should encode"), + ) + .await + .expect("newer legacy backup should persist"); + + let err = persisted_usage_floor_for_startup(store, false) + .await + .expect_err("newer legacy backup must not be hidden by an older bootstrap marker"); + assert!( + err.to_string() + .contains("scanner usage bootstrap conflicts with a persisted backup"), + "unexpected conflict error: {err}" + ); +} + +#[tokio::test] +async fn scanner_usage_state_reset_slots_reject_primary_aba() { + let store = Arc::new(MemoryConfigStore::default()); + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); + store.objects.lock().await.insert(key.clone(), b"not-json".to_vec()); + store.revisions.lock().await.insert(key.clone(), 1); + let slots = read_usage_state_reset_slots(store.clone()) + .await + .expect("usage reset slots should be inspected"); + + store.objects.lock().await.insert(key.clone(), b"newer-json".to_vec()); + store.revisions.lock().await.insert(key, 2); + let err = reset_scanner_usage_state_slots_for_full_rebuild(store, &slots, 0, 3) + .await + .expect_err("stale primary revision must not be overwritten"); + assert!( + err.to_string() + .contains("scanner usage reset primary slot changed before bootstrap publish"), + "unexpected primary ABA error: {err}" + ); +} + +#[tokio::test] +async fn scanner_usage_state_reset_slots_defer_when_publication_epoch_moves() { + let store = Arc::new(MemoryConfigStore::default()); + let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); + store.objects.lock().await.insert(key, b"not-json".to_vec()); + let slots = read_usage_state_reset_slots(store.clone()) + .await + .expect("usage reset slots should be inspected"); + store.publication_admission_blocked.store(true, Ordering::Release); + + let err = reset_scanner_usage_state_slots_for_full_rebuild(store, &slots, 0, 3) + .await + .expect_err("movement admission loss must defer reset"); + assert!( + err.to_string() + .contains("scanner usage reset deferred by a movement epoch change"), + "unexpected movement defer error: {err}" + ); +} + #[tokio::test] async fn leadership_claim_defers_on_corrupt_usage_baseline_without_bloom_write() { let store = Arc::new(MemoryConfigStore::default()); diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index 5242db647..aade3999b 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -187,6 +187,57 @@ catch_up_estimate.discovered_expiry_items catch_up_estimate.discovered_transition_items ``` +## Usage State Reset + +The supported break-glass route for rebuilding scanner usage state is: + +```text +POST /v3/scanner/usage-state/reset +{"mode":"full-rebuild"} +``` + +The request must be authenticated with an admin identity that has +`ConfigUpdateAdminAction`. Use it only after confirming the scanner status shows +a usage-floor load failure, a conflicting persisted usage floor, or an operator +decision to discard the durable usage baseline and rebuild it from a full +scanner pass. + +The reset does not delete metadata files from disk by hand and does not publish +an authoritative zero-usage snapshot. It holds the scanner leader lock, fences +the operation with the storage-owned publication epoch, CAS-publishes a v2 +`bootstrap-pending` marker in the primary usage slot, then clears stale backup, +legacy, and observed usage slots by object revision. The next scanner +leadership claim binds that marker to a fresh epoch and the next complete +scanner cycle replaces it with authoritative usage. + +The JSON response is machine-readable: + +```json +{ + "status": "reset", + "mode": "full-rebuild", + "usage_state": "bootstrap-pending", + "leader_epoch": 9, + "next_cycle": 42, + "reset_paths": [ + "buckets/.usage.v2.json", + "buckets/.usage.v2.json.bkp", + "buckets/.usage.json", + "buckets/.usage.json.bkp", + "buckets/.usage.observed.json" + ] +} +``` + +If the response is an error mentioning data movement, wait for decommission or +rebalance to leave the scanner metadata path and retry. If it reports that the +scanner cycle state is invalid, run the cycle-state recovery reset first: + +```text +POST /v3/scanner/cycle-state/reset +{"mode":"full-rescan"} +``` + ## Data Movement Pauses RustFS currently uses a `global_pause` policy while pool decommission or diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index a18b579d5..024b2676e 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -152,6 +152,7 @@ mod tests { let _remove_remote_target_handler = replication::RemoveRemoteTargetHandler {}; let _scanner_status_handler = scanner::ScannerStatusHandler {}; let _scanner_cycle_state_reset_handler = scanner::ScannerCycleStateResetHandler {}; + let _scanner_usage_state_reset_handler = scanner::ScannerUsageStateResetHandler {}; 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 47815ae1e..638133479 100644 --- a/rustfs/src/admin/handlers/scanner.rs +++ b/rustfs/src/admin/handlers/scanner.rs @@ -60,6 +60,12 @@ struct ScannerCycleResetRequest { mode: String, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct ScannerUsageStateResetRequest { + mode: String, +} + #[derive(Debug, Serialize)] struct ScannerFreshnessStatus { state: &'static str, @@ -233,6 +239,11 @@ pub fn register_scanner_route(r: &mut S3Router) -> std::io::Resu format!("{ADMIN_PREFIX}/v3/scanner/cycle-state/reset").as_str(), AdminOperation(&ScannerCycleStateResetHandler {}), )?; + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}/v3/scanner/usage-state/reset").as_str(), + AdminOperation(&ScannerUsageStateResetHandler {}), + )?; r.insert( Method::GET, format!("{ADMIN_PREFIX}/v3/ilm/expiry/status").as_str(), @@ -303,6 +314,8 @@ pub struct IlmExpiryStatusHandler {} pub struct ScannerCycleStateResetHandler {} +pub struct ScannerUsageStateResetHandler {} + #[async_trait::async_trait] impl Operation for ScannerCycleStateResetHandler { async fn call(&self, mut req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -332,6 +345,40 @@ impl Operation for ScannerCycleStateResetHandler { } } +#[async_trait::async_trait] +impl Operation for ScannerUsageStateResetHandler { + 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-rebuild" { + return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "reset mode must be full-rebuild")); + } + 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"))?; + let result = supervise_admin_mutation("scanner usage state reset", async move { + rustfs_scanner::scanner::reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), store) + .await + .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string())) + }) + .await?; + let body = serde_json::to_vec(&result).map_err(|err| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("failed to encode scanner usage reset response: {err}"), + ) + })?; + json_response(body) + } +} + #[async_trait::async_trait] impl Operation for IlmExpiryStatusHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -422,6 +469,20 @@ mod tests { assert!(serde_json::from_str::(r#"{"mode":"full-rescan","cursor":"untrusted"}"#).is_err()); } + #[test] + fn admin_usage_reset_requires_full_rebuild_without_untrusted_fields() { + let full_rebuild: ScannerUsageStateResetRequest = + serde_json::from_str(r#"{"mode":"full-rebuild"}"#).expect("full rebuild must be accepted"); + assert_eq!(full_rebuild.mode, "full-rebuild"); + let cycle_mode: ScannerUsageStateResetRequest = + serde_json::from_str(r#"{"mode":"full-rescan"}"#).expect("mode validation belongs to the handler"); + assert_ne!(cycle_mode.mode, "full-rebuild"); + assert!( + serde_json::from_str::(r#"{"mode":"full-rebuild","delete_files":[".usage.v2.json"]}"#) + .is_err() + ); + } + #[test] fn scanner_disabled_reason_reports_startup_env_key() { assert_eq!(scanner_disabled_reason(true), None); diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 850a432d4..c6d94a9f6 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -445,6 +445,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ CONFIG_UPDATE, RouteRiskLevel::High, ), + admin( + HttpMethod::Post, + "/rustfs/admin/v3/scanner/usage-state/reset", + CONFIG_UPDATE, + RouteRiskLevel::High, + ), admin( HttpMethod::Get, "/rustfs/admin/v3/ilm/expiry/status", @@ -2107,6 +2113,12 @@ mod tests { assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", SERVER_INFO); } + #[test] + fn route_policy_requires_config_update_for_scanner_usage_reset() { + assert_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/usage-state/reset", CONFIG_UPDATE); + assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/usage-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 d8e20620d..9ad2a06de 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -255,6 +255,7 @@ fn expected_admin_route_matrix() -> Vec { 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::POST, "/v3/scanner/usage-state/reset"), admin_route(Method::GET, "/v3/audit/target/list"), admin_route_sample( Method::PUT, @@ -909,6 +910,7 @@ fn test_register_routes_cover_representative_admin_paths() { 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::POST, &admin_path("/v3/scanner/usage-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( @@ -1399,6 +1401,7 @@ fn test_admin_alias_paths_match_existing_admin_routes() { (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::POST, compat_admin_alias_path("/v3/scanner/usage-state/reset")), (Method::GET, compat_admin_alias_path("/v3/ilm/expiry/status")), ] { assert!(