From 21015cfac8ba728eff06019989cf660c316a41da Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 12:54:04 +0800 Subject: [PATCH] feat(scanner): accept durable recovery intents (#7344) * feat(scanner): accept durable recovery intents Add a CAS-backed scanner usage recovery intent record for async full rebuild admission. The admin reset endpoint can now persist and replay idempotent intent acceptance before returning 202, and a read-only status route exposes the durable request state. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * feat(madmin): add scanner recovery intent helpers (#7345) Expose madmin helpers for accepting and querying asynchronous scanner usage-state full-rebuild recovery intents. Keep the legacy synchronous helper unchanged and pin the new request/response wire contract with focused client tests. Co-authored-by: zhi22915 * feat(scanner): execute recovery intents asynchronously Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- crates/madmin/src/client.rs | 121 +++++- crates/scanner/src/lib.rs | 11 +- crates/scanner/src/scanner.rs | 7 +- crates/scanner/src/scanner/cycle_state.rs | 365 ++++++++++++++++++ .../src/scanner/tests/recovery_control.rs | 196 ++++++++++ rustfs/src/admin/handlers/scanner.rs | 265 ++++++++++++- rustfs/src/admin/route_policy.rs | 20 + rustfs/src/admin/route_registration_test.rs | 16 + 8 files changed, 993 insertions(+), 8 deletions(-) diff --git a/crates/madmin/src/client.rs b/crates/madmin/src/client.rs index ac9bebfc2..f76c1048f 100644 --- a/crates/madmin/src/client.rs +++ b/crates/madmin/src/client.rs @@ -254,6 +254,19 @@ pub struct ScannerUsageStateResetResponse { pub extra: serde_json::Map, } +/// Durable scanner usage-state recovery intent response. +#[derive(Debug, Clone, Deserialize)] +pub struct ScannerUsageRecoveryIntentResponse { + /// `accepted`, `replayed`, or `found`. + pub status: String, + pub action: String, + pub mode: String, + pub intent_id: String, + pub state: String, + #[serde(flatten)] + pub extra: serde_json::Map, +} + #[derive(Debug, Serialize)] struct ScannerCycleResetRequest<'a> { mode: &'a str, @@ -264,6 +277,14 @@ struct ScannerUsageStateResetRequest<'a> { mode: &'a str, } +#[derive(Debug, Serialize)] +struct ScannerUsageStateAsyncResetRequest<'a> { + mode: &'a str, + #[serde(rename = "async")] + async_intent: bool, + idempotency_key: &'a str, +} + /// Freshness block of the scanner status response. #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] @@ -472,6 +493,38 @@ impl AdminClient { self.post_json("/v3/scanner/usage-state/reset", &[], body).await } + /// Accept a durable asynchronous scanner usage-state full-rebuild intent. + /// + /// The caller owns `idempotency_key`; replaying the same key on the same + /// server-side actor returns the same accepted intent instead of starting + /// the legacy synchronous reset path. + pub async fn scanner_usage_state_accept_full_rebuild_intent( + &self, + idempotency_key: &str, + ) -> Result { + let body = serde_json::to_vec(&ScannerUsageStateAsyncResetRequest { + mode: "full-rebuild", + async_intent: true, + idempotency_key, + }) + .map_err(|err| AdminClientError::Decode { + message: err.to_string(), + })?; + self.post_json("/v3/scanner/usage-state/reset", &[], body).await + } + + /// Query a durable asynchronous scanner usage-state recovery intent. + pub async fn scanner_usage_state_recovery_intent_status( + &self, + intent_id: &str, + ) -> Result { + self.get_json(&format!( + "/v3/scanner/usage-state/recovery-intents/{}", + percent_encode_path_segment(intent_id) + )) + .await + } + /// ILM expiry worker status. The payload is owned by the expiry /// subsystem and still evolving; returned verbatim. pub async fn ilm_expiry_status(&self) -> Result { @@ -640,7 +693,8 @@ pub(crate) fn percent_encode_path_segment(segment: &str) -> String { mod tests { use super::{ AdminClient, AdminClientError, BackgroundHealStatus, HealOpts, HealScanMode, HealStartSuccess, HealTaskStatus, - ScannerCycleResetResponse, ScannerStatus, ScannerUsageStateResetResponse, heal_path, percent_encode_path_segment, + ScannerCycleResetResponse, ScannerStatus, ScannerUsageRecoveryIntentResponse, ScannerUsageStateResetResponse, heal_path, + percent_encode_path_segment, }; use crate::test_support::TestServer; use serde_json::json; @@ -808,6 +862,22 @@ mod tests { assert_eq!(usage.next_cycle, 42); assert_eq!(usage.reset_paths, [".usage.json"]); assert_eq!(usage.extra["future"]["accepted"], false); + + let intent: ScannerUsageRecoveryIntentResponse = serde_json::from_value(json!({ + "status": "accepted", + "action": "usage-full-rebuild", + "mode": "full-rebuild", + "intent_id": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + "state": "accepted", + "future": {"worker": "pending"} + })) + .unwrap(); + assert_eq!(intent.status, "accepted"); + assert_eq!(intent.action, "usage-full-rebuild"); + assert_eq!(intent.mode, "full-rebuild"); + assert_eq!(intent.intent_id, "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"); + assert_eq!(intent.state, "accepted"); + assert_eq!(intent.extra["future"]["worker"], "pending"); } #[test] @@ -898,6 +968,55 @@ mod tests { assert!(request.body.contains("\"mode\":\"full-rebuild\"")); } + #[tokio::test] + async fn scanner_usage_async_reset_posts_explicit_intent_contract() { + let server = TestServer::spawn( + r#"{"status":"accepted","action":"usage-full-rebuild","mode":"full-rebuild","intent_id":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","state":"accepted"}"#, + 202, + ) + .await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + + let accepted = client + .scanner_usage_state_accept_full_rebuild_intent("intent-key-1") + .await + .expect("async recovery intent response decodes"); + + assert_eq!(accepted.status, "accepted"); + assert_eq!(accepted.mode, "full-rebuild"); + assert_eq!(accepted.state, "accepted"); + let request = server.recorded(); + assert_eq!(request.method, "POST"); + assert_eq!(request.path, "/rustfs/admin/v3/scanner/usage-state/reset"); + assert_eq!(request.query, ""); + assert!(request.body.contains("\"mode\":\"full-rebuild\"")); + assert!(request.body.contains("\"async\":true")); + assert!(request.body.contains("\"idempotency_key\":\"intent-key-1\"")); + } + + #[tokio::test] + async fn scanner_usage_recovery_intent_status_gets_encoded_intent_id() { + let server = TestServer::spawn( + r#"{"status":"found","action":"usage-full-rebuild","mode":"full-rebuild","intent_id":"id%2Fwith%20space","state":"accepted"}"#, + 200, + ) + .await; + let client = AdminClient::new(&format!("http://{}", server.addr), "ak", "sk").unwrap(); + + let status = client + .scanner_usage_state_recovery_intent_status("id/with space") + .await + .expect("recovery intent status response decodes"); + + assert_eq!(status.status, "found"); + assert_eq!(status.action, "usage-full-rebuild"); + let request = server.recorded(); + assert_eq!(request.method, "GET"); + assert_eq!(request.path, "/rustfs/admin/v3/scanner/usage-state/recovery-intents/id%2Fwith%20space"); + assert_eq!(request.query, ""); + assert_eq!(request.body, ""); + } + #[tokio::test] async fn query_sends_client_token_on_the_same_path() { let body = r#"{"summary":"running","detail":"","settings":{"recursive":false},"items":[],"truncated":false}"#; diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 3123b66f9..cfc63a3b7 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -83,10 +83,13 @@ pub use remote_scanner::{ pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_status, validate_scanner_runtime_config}; pub use rustfs_scanner_metrics::last_minute; pub use scanner::{ - ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, - ScannerPauseBacklogPhase, ScannerPauseBacklogStatus, ScannerPauseBacklogThresholds, ScannerUsageStateResetResult, - init_data_scanner, init_scanner_with_recovery, reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, - scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_topology_digest, + SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD, ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, + ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogStatus, + ScannerPauseBacklogThresholds, ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict, ScannerRecoveryIntentRecord, + ScannerRecoveryIntentRequest, ScannerUsageStateResetResult, accept_scanner_usage_recovery_intent, + get_scanner_usage_recovery_intent, init_data_scanner, init_scanner_with_recovery, reset_scanner_cycle_recovery, + reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent, scanner_cycle_recovery_status, + scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_recovery_actor_sha256, scanner_topology_digest, }; pub use scanner_io::{ ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState, diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 03f87a61f..f2337c3ea 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -3590,8 +3590,11 @@ pub use backlog::{ #[cfg(test)] pub(crate) use cycle_state::encode_scanner_cycle_fence_for_test; pub use cycle_state::{ - ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerUsageStateResetResult, reset_scanner_cycle_recovery, - reset_scanner_usage_state_for_full_rebuild, scanner_cycle_recovery_status, + SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD, ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, + ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict, ScannerRecoveryIntentRecord, ScannerRecoveryIntentRequest, + ScannerUsageStateResetResult, accept_scanner_usage_recovery_intent, get_scanner_usage_recovery_intent, + reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent, + scanner_cycle_recovery_status, scanner_recovery_actor_sha256, }; 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 dfa9b8cdf..646fbeead 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -34,6 +34,14 @@ const LEGACY_INCOMPLETE_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"; +const SCANNER_RECOVERY_INTENT_SCHEMA_VERSION: u16 = 1; +const SCANNER_RECOVERY_INTENT_PREFIX: &str = ".usage.v2.recovery-intents"; +const MAX_SCANNER_RECOVERY_INTENT_BYTES: u64 = 16 * 1024; +const SCANNER_RECOVERY_INTENT_STATE_ACCEPTED: &str = "accepted"; +const SCANNER_RECOVERY_INTENT_STATE_RUNNING: &str = "running"; +const SCANNER_RECOVERY_INTENT_STATE_COMPLETED: &str = "completed"; +const SCANNER_RECOVERY_INTENT_STATE_FAILED: &str = "failed"; +pub const SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD: &str = "scanner-usage-full-rebuild"; #[cfg(test)] pub(super) mod cleanup_io_fault { @@ -146,6 +154,43 @@ pub struct ScannerUsageStateResetResult { pub reset_paths: Vec, } +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct ScannerRecoveryIntentRequest { + pub action: String, + pub mode: String, + pub idempotency_key: String, + pub actor_sha256: String, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct ScannerRecoveryIntentRecord { + pub schema_version: u16, + pub intent_id: String, + pub action: String, + pub mode: String, + pub state: String, + pub actor_sha256: String, + pub idempotency_key_sha256: String, + pub request_sha256: String, + pub accepted_at_unix_secs: u64, +} + +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +pub struct ScannerRecoveryIntentConflict { + pub intent_id: String, + pub state: String, +} + +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] +#[serde(tag = "outcome", rename_all = "snake_case")] +pub enum ScannerRecoveryIntentAcceptResult { + Accepted { record: ScannerRecoveryIntentRecord }, + Replayed { record: ScannerRecoveryIntentRecord }, + Conflict { existing: ScannerRecoveryIntentConflict }, +} + #[derive(Clone, Debug)] pub(super) struct ScannerUsageStateResetSlot { path: String, @@ -360,6 +405,326 @@ fn unix_now_secs() -> u64 { u64::try_from(Utc::now().timestamp()).unwrap_or(0) } +fn sha256_hex(parts: &[&[u8]]) -> String { + let mut hasher = Sha256::new(); + for part in parts { + hasher.update(part); + hasher.update([0]); + } + const HEX: &[u8; 16] = b"0123456789abcdef"; + let digest = hasher.finalize(); + let mut encoded = String::with_capacity(64); + for byte in digest { + encoded.push(char::from(HEX[usize::from(byte >> 4)])); + encoded.push(char::from(HEX[usize::from(byte & 0x0f)])); + } + encoded +} + +fn is_canonical_sha256(value: &str) -> bool { + value.len() == 64 + && value + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) +} + +fn validate_idempotency_key(key: &str) -> Result<(), ScannerError> { + if !(8..=256).contains(&key.len()) { + return Err(ScannerError::Other( + "scanner recovery intent idempotency key length is unsupported".to_string(), + )); + } + if !key + .bytes() + .all(|byte| byte.is_ascii() && !byte.is_ascii_control() && !byte.is_ascii_whitespace()) + { + return Err(ScannerError::Other( + "scanner recovery intent idempotency key must be printable ASCII without whitespace".to_string(), + )); + } + Ok(()) +} + +fn validate_recovery_intent_request(request: &ScannerRecoveryIntentRequest) -> Result<(), ScannerError> { + if request.action != SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD { + return Err(ScannerError::Other("scanner recovery intent action is unsupported".to_string())); + } + if request.mode != SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD { + return Err(ScannerError::Other("scanner recovery intent mode is unsupported".to_string())); + } + if !is_canonical_sha256(&request.actor_sha256) { + return Err(ScannerError::Other("scanner recovery intent actor identity is invalid".to_string())); + } + validate_idempotency_key(&request.idempotency_key) +} + +fn scanner_recovery_intent_path(intent_id: &str) -> Result { + if !is_canonical_sha256(intent_id) { + return Err(ScannerError::Other("scanner recovery intent id is invalid".to_string())); + } + Ok(format!("{SCANNER_RECOVERY_INTENT_PREFIX}/{intent_id}.json")) +} + +pub fn scanner_recovery_actor_sha256(actor: &str) -> String { + sha256_hex(&[b"scanner-recovery-actor-v1", actor.as_bytes()]) +} + +fn scanner_recovery_intent_candidate( + request: &ScannerRecoveryIntentRequest, +) -> Result { + validate_recovery_intent_request(request)?; + let intent_id = sha256_hex(&[ + b"scanner-recovery-intent-v1", + request.actor_sha256.as_bytes(), + request.idempotency_key.as_bytes(), + ]); + Ok(ScannerRecoveryIntentRecord { + schema_version: SCANNER_RECOVERY_INTENT_SCHEMA_VERSION, + intent_id, + action: request.action.clone(), + mode: request.mode.clone(), + state: SCANNER_RECOVERY_INTENT_STATE_ACCEPTED.to_string(), + actor_sha256: request.actor_sha256.clone(), + idempotency_key_sha256: sha256_hex(&[b"scanner-recovery-idempotency-key-v1", request.idempotency_key.as_bytes()]), + request_sha256: sha256_hex(&[ + b"scanner-recovery-request-v1", + request.actor_sha256.as_bytes(), + request.action.as_bytes(), + request.mode.as_bytes(), + request.idempotency_key.as_bytes(), + ]), + accepted_at_unix_secs: unix_now_secs(), + }) +} + +fn validate_recovery_intent_record(record: &ScannerRecoveryIntentRecord) -> Result<(), ScannerError> { + if record.schema_version != SCANNER_RECOVERY_INTENT_SCHEMA_VERSION { + return Err(ScannerError::Other("scanner recovery intent schema is unsupported".to_string())); + } + if !is_canonical_sha256(&record.intent_id) + || !is_canonical_sha256(&record.actor_sha256) + || !is_canonical_sha256(&record.idempotency_key_sha256) + || !is_canonical_sha256(&record.request_sha256) + { + return Err(ScannerError::Other("scanner recovery intent identity is invalid".to_string())); + } + if record.action != SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD + || record.mode != SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD + || !matches!( + record.state.as_str(), + SCANNER_RECOVERY_INTENT_STATE_ACCEPTED + | SCANNER_RECOVERY_INTENT_STATE_RUNNING + | SCANNER_RECOVERY_INTENT_STATE_COMPLETED + | SCANNER_RECOVERY_INTENT_STATE_FAILED + ) + { + return Err(ScannerError::Other("scanner recovery intent state is invalid".to_string())); + } + Ok(()) +} + +fn decode_recovery_intent_record(data: &[u8]) -> Result { + let record: ScannerRecoveryIntentRecord = + serde_json::from_slice(data).map_err(|err| ScannerError::Other(format!("scanner recovery intent is invalid: {err}")))?; + validate_recovery_intent_record(&record)?; + Ok(record) +} + +fn compare_recovery_intent( + expected: &ScannerRecoveryIntentRecord, + existing: ScannerRecoveryIntentRecord, +) -> ScannerRecoveryIntentAcceptResult { + if existing.actor_sha256 == expected.actor_sha256 + && existing.action == expected.action + && existing.mode == expected.mode + && existing.idempotency_key_sha256 == expected.idempotency_key_sha256 + && existing.request_sha256 == expected.request_sha256 + { + ScannerRecoveryIntentAcceptResult::Replayed { record: existing } + } else { + ScannerRecoveryIntentAcceptResult::Conflict { + existing: ScannerRecoveryIntentConflict { + intent_id: existing.intent_id, + state: existing.state, + }, + } + } +} + +async fn read_recovery_intent_record( + storeapi: Arc, + path: &str, +) -> Result, ScannerError> { + read_recovery_intent_record_with_revision(storeapi, path) + .await + .map(|record| record.map(|(record, _)| record)) +} + +async fn read_recovery_intent_record_with_revision( + storeapi: Arc, + path: &str, +) -> Result, ScannerError> { + let mut reader = match storeapi + .get_object_reader( + RUSTFS_META_BUCKET, + path, + 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), + Err(err) => return Err(ScannerError::Other(format!("failed to read scanner recovery intent: {err}"))), + }; + let revision = reader + .object_info + .etag + .as_ref() + .filter(|etag| !etag.is_empty()) + .cloned() + .map(DataUsageCacheRevision::Etag) + .ok_or_else(|| ScannerError::Other("scanner recovery intent has no revision".to_string()))?; + let max_object_size = i64::try_from(MAX_SCANNER_RECOVERY_INTENT_BYTES).unwrap_or(i64::MAX); + if reader.object_info.is_dir || reader.object_info.size < 0 || reader.object_info.size > max_object_size { + return Err(ScannerError::Other("scanner recovery intent exceeds the bounded object size".to_string())); + } + let max_len = usize::try_from(MAX_SCANNER_RECOVERY_INTENT_BYTES).unwrap_or(usize::MAX); + let mut data = Vec::new(); + (&mut reader) + .take(MAX_SCANNER_RECOVERY_INTENT_BYTES.saturating_add(1)) + .read_to_end(&mut data) + .await + .map_err(|err| ScannerError::Other(format!("failed to read scanner recovery intent: {err}")))?; + if data.is_empty() { + return Err(ScannerError::Other("scanner recovery intent is empty".to_string())); + } + if data.len() > max_len { + return Err(ScannerError::Other("scanner recovery intent exceeds the bounded object size".to_string())); + } + decode_recovery_intent_record(&data).map(|record| Some((record, revision))) +} + +pub async fn get_scanner_usage_recovery_intent( + storeapi: Arc, + intent_id: &str, +) -> Result, ScannerError> { + let path = scanner_recovery_intent_path(intent_id)?; + read_recovery_intent_record(storeapi, &path).await +} + +pub async fn accept_scanner_usage_recovery_intent( + storeapi: Arc, + request: ScannerRecoveryIntentRequest, +) -> Result { + let candidate = scanner_recovery_intent_candidate(&request)?; + let path = scanner_recovery_intent_path(&candidate.intent_id)?; + let encoded = serde_json::to_vec(&candidate) + .map_err(|err| ScannerError::Other(format!("failed to encode scanner recovery intent: {err}")))?; + match save_config_with_preconditions(storeapi.clone(), &path, encoded, DataUsageCacheRevision::Missing.preconditions()).await + { + Ok(_) => Ok(ScannerRecoveryIntentAcceptResult::Accepted { record: candidate }), + Err(EcstoreError::PreconditionFailed) => { + let existing = read_recovery_intent_record(storeapi, &path).await?; + let Some(existing) = existing else { + return Err(ScannerError::Other( + "scanner recovery intent disappeared after creation conflict".to_string(), + )); + }; + Ok(compare_recovery_intent(&candidate, existing)) + } + Err(err) => Err(ScannerError::Other(format!("failed to persist scanner recovery intent: {err}"))), + } +} + +async fn transition_scanner_usage_recovery_intent( + storeapi: Arc, + intent_id: &str, + expected_states: &[&str], + next_state: &str, +) -> Result, ScannerError> { + let path = scanner_recovery_intent_path(intent_id)?; + for _ in 0..8 { + let Some((mut record, revision)) = read_recovery_intent_record_with_revision(storeapi.clone(), &path).await? else { + return Err(ScannerError::Other("scanner recovery intent disappeared before execution".to_string())); + }; + if record.state == next_state { + return Ok(Some(record)); + } + if !expected_states.contains(&record.state.as_str()) { + return Ok(None); + } + record.state = next_state.to_string(); + let encoded = serde_json::to_vec(&record) + .map_err(|err| ScannerError::Other(format!("failed to encode scanner recovery intent: {err}")))?; + match save_config_with_preconditions(storeapi.clone(), &path, encoded, revision.preconditions()).await { + Ok(_) => return Ok(Some(record)), + Err(EcstoreError::PreconditionFailed) => continue, + Err(err) => return Err(ScannerError::Other(format!("failed to persist scanner recovery intent: {err}"))), + } + } + Err(ScannerError::Other( + "scanner recovery intent state changed repeatedly during transition".to_string(), + )) +} + +pub async fn run_scanner_usage_recovery_intent( + ctx: CancellationToken, + storeapi: Arc, + intent_id: String, +) -> Result, ScannerError> { + let Some(record) = transition_scanner_usage_recovery_intent( + storeapi.clone(), + &intent_id, + &[SCANNER_RECOVERY_INTENT_STATE_ACCEPTED, SCANNER_RECOVERY_INTENT_STATE_RUNNING], + SCANNER_RECOVERY_INTENT_STATE_RUNNING, + ) + .await? + else { + return Ok(None); + }; + if record.action != SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD + || record.mode != SCANNER_USAGE_STATE_RESET_MODE_FULL_REBUILD + { + return Err(ScannerError::Other( + "scanner recovery intent action or mode changed before execution".to_string(), + )); + } + + match reset_scanner_usage_state_for_full_rebuild(ctx, storeapi.clone()).await { + Ok(result) => { + transition_scanner_usage_recovery_intent( + storeapi, + &intent_id, + &[SCANNER_RECOVERY_INTENT_STATE_RUNNING, SCANNER_RECOVERY_INTENT_STATE_FAILED], + SCANNER_RECOVERY_INTENT_STATE_COMPLETED, + ) + .await?; + Ok(Some(result)) + } + Err(err) => { + let _ = transition_scanner_usage_recovery_intent( + storeapi, + &intent_id, + &[SCANNER_RECOVERY_INTENT_STATE_RUNNING], + SCANNER_RECOVERY_INTENT_STATE_FAILED, + ) + .await; + Err(err) + } + } +} + fn recovery_status(state: &str, reason: Option<&str>, retryable: bool) -> ScannerCycleRecoveryStatus { ScannerCycleRecoveryStatus { path: DATA_USAGE_BLOOM_NAME_PATH.clone(), diff --git a/crates/scanner/src/scanner/tests/recovery_control.rs b/crates/scanner/src/scanner/tests/recovery_control.rs index 7f0ba515a..5f927011a 100644 --- a/crates/scanner/src/scanner/tests/recovery_control.rs +++ b/crates/scanner/src/scanner/tests/recovery_control.rs @@ -97,6 +97,15 @@ async fn run_disabled_startup(ctx: CancellationToken, store: Arc) { ); } +fn recovery_intent_request(key: &str, actor: &str) -> ScannerRecoveryIntentRequest { + ScannerRecoveryIntentRequest { + action: SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD.to_string(), + mode: "full-rebuild".to_string(), + idempotency_key: key.to_string(), + actor_sha256: scanner_recovery_actor_sha256(actor), + } +} + async fn assert_reset_fences(store: &Arc) { let data = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()) .await @@ -203,6 +212,193 @@ async fn disabled_cleanup_recovers_after_child_process_crash_boundaries() { } } +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_accept_is_durable_and_idempotent() { + let (_dir, store) = setup_scanner_cycle_store().await; + let request = recovery_intent_request("intent-key-0001", "operator-a"); + + let first = accept_scanner_usage_recovery_intent(store.clone(), request.clone()) + .await + .expect("first intent should persist"); + let record = match first { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create a durable intent: {other:?}"), + }; + assert_eq!(record.state, "accepted"); + assert_eq!(record.action, SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD); + assert_ne!(record.idempotency_key_sha256, request.idempotency_key); + + let restarted = restart_scanner_cycle_store_from(&store).await; + let queried = get_scanner_usage_recovery_intent(restarted.clone(), &record.intent_id) + .await + .expect("persisted intent should read after restart") + .expect("persisted intent should exist"); + assert_eq!(queried, record); + + let replay = accept_scanner_usage_recovery_intent(restarted, request) + .await + .expect("lost response retry should be idempotent"); + assert_eq!(replay, ScannerRecoveryIntentAcceptResult::Replayed { record }); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_executor_persists_completed_progress() { + let (_dir, store) = setup_scanner_cycle_store().await; + let request = recovery_intent_request("intent-key-0001-exec", "operator-a"); + + let record = match accept_scanner_usage_recovery_intent(store.clone(), request.clone()) + .await + .expect("intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create a durable intent: {other:?}"), + }; + + let reset = run_scanner_usage_recovery_intent(CancellationToken::new(), store.clone(), record.intent_id.clone()) + .await + .expect("accepted intent should execute") + .expect("accepted intent should produce a reset"); + assert_eq!(reset.status, "reset"); + assert_eq!(reset.mode, "full-rebuild"); + + let completed = get_scanner_usage_recovery_intent(store.clone(), &record.intent_id) + .await + .expect("completed intent should read") + .expect("completed intent should remain durable"); + assert_eq!(completed.state, "completed"); + assert_eq!(completed.intent_id, record.intent_id); + + let restarted = restart_scanner_cycle_store_from(&store).await; + let replay = accept_scanner_usage_recovery_intent(restarted.clone(), request) + .await + .expect("lost response retry should return the durable completed intent"); + assert_eq!(replay, ScannerRecoveryIntentAcceptResult::Replayed { record: completed }); + let rerun = run_scanner_usage_recovery_intent(CancellationToken::new(), restarted, record.intent_id) + .await + .expect("terminal intent should not be an execution error"); + assert!(rerun.is_none(), "completed intent must not start a second reset"); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_executor_persists_failed_progress() { + let (_dir, store) = setup_scanner_cycle_store().await; + let record = match accept_scanner_usage_recovery_intent( + store.clone(), + recovery_intent_request("intent-key-0001-failed", "operator-a"), + ) + .await + .expect("intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create a durable intent: {other:?}"), + }; + + let ctx = CancellationToken::new(); + ctx.cancel(); + let error = run_scanner_usage_recovery_intent(ctx, store.clone(), record.intent_id.clone()) + .await + .expect_err("cancelled intent execution should fail"); + assert!(error.to_string().contains("cancelled"), "{error}"); + + let failed = get_scanner_usage_recovery_intent(store, &record.intent_id) + .await + .expect("failed intent should read") + .expect("failed intent should remain durable"); + assert_eq!(failed.state, "failed"); + assert_eq!(failed.intent_id, record.intent_id); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_rejects_same_namespace_conflict() { + let (_dir, store) = setup_scanner_cycle_store().await; + let request = recovery_intent_request("intent-key-0002", "operator-a"); + let record = match accept_scanner_usage_recovery_intent(store.clone(), request.clone()) + .await + .expect("first intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create a durable intent: {other:?}"), + }; + + let mut unsupported = request.clone(); + unsupported.mode = "future-mode".to_string(); + let error = accept_scanner_usage_recovery_intent(store.clone(), unsupported) + .await + .expect_err("unsupported mode must fail before it can collide with a durable record"); + assert!(error.to_string().contains("mode is unsupported")); + + let same_key_other_actor = recovery_intent_request("intent-key-0002", "operator-b"); + let accepted_other_actor = accept_scanner_usage_recovery_intent(store.clone(), same_key_other_actor) + .await + .expect("another actor owns an independent idempotency namespace"); + assert!(matches!(accepted_other_actor, ScannerRecoveryIntentAcceptResult::Accepted { .. })); + + let path = format!(".usage.v2.recovery-intents/{}.json", record.intent_id); + let mut existing = record.clone(); + existing.request_sha256 = scanner_recovery_actor_sha256("different-request"); + save_config(store.clone(), &path, serde_json::to_vec(&existing).expect("mutated record should encode")) + .await + .expect("mutate durable record"); + let conflict = accept_scanner_usage_recovery_intent(store, request) + .await + .expect("same namespace collision should be reported"); + assert_eq!( + conflict, + ScannerRecoveryIntentAcceptResult::Conflict { + existing: ScannerRecoveryIntentConflict { + intent_id: record.intent_id, + state: "accepted".to_string(), + }, + } + ); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_query_rejects_corrupt_or_unknown_records() { + let (_dir, store) = setup_scanner_cycle_store().await; + let request = recovery_intent_request("intent-key-0003", "operator-a"); + let record = match accept_scanner_usage_recovery_intent(store.clone(), request) + .await + .expect("first intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create a durable intent: {other:?}"), + }; + let path = format!(".usage.v2.recovery-intents/{}.json", record.intent_id); + save_config(store.clone(), &path, b"{corrupt".to_vec()) + .await + .expect("corrupt durable record"); + let error = get_scanner_usage_recovery_intent(store.clone(), &record.intent_id) + .await + .expect_err("corrupt intent must not decode as absent"); + assert!(error.to_string().contains("scanner recovery intent is invalid")); + let unknown = get_scanner_usage_recovery_intent(store, &scanner_recovery_actor_sha256("missing")) + .await + .expect("missing intent should read as absent"); + assert!(unknown.is_none()); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_query_rejects_oversized_records() { + let (_dir, store) = setup_scanner_cycle_store().await; + let intent_id = scanner_recovery_actor_sha256("oversized-intent"); + let path = format!(".usage.v2.recovery-intents/{intent_id}.json"); + save_config(store.clone(), &path, vec![b'a'; 16 * 1024 + 1]) + .await + .expect("oversized durable record"); + + let error = get_scanner_usage_recovery_intent(store, &intent_id) + .await + .expect_err("oversized intent must not be materialized"); + assert!(error.to_string().contains("bounded object size"), "{error}"); +} + #[tokio::test] #[serial] async fn disabled_cleanup_reopens_persisted_intent_without_starting_scanner() { diff --git a/rustfs/src/admin/handlers/scanner.rs b/rustfs/src/admin/handlers/scanner.rs index 077323463..f96f71406 100644 --- a/rustfs/src/admin/handlers/scanner.rs +++ b/rustfs/src/admin/handlers/scanner.rs @@ -62,6 +62,26 @@ struct ScannerCycleResetRequest { #[serde(deny_unknown_fields)] struct ScannerUsageStateResetRequest { mode: String, + #[serde(default, rename = "async")] + async_intent: bool, + #[serde(default)] + idempotency_key: Option, +} + +#[derive(Debug, Serialize)] +struct ScannerRecoveryIntentResponse { + status: &'static str, + action: String, + mode: String, + intent_id: String, + state: String, +} + +#[derive(Debug, Serialize)] +struct ScannerRecoveryIntentConflictResponse { + status: &'static str, + intent_id: String, + state: String, } #[derive(Debug, Serialize)] @@ -242,6 +262,11 @@ pub fn register_scanner_route(r: &mut S3Router) -> std::io::Resu format!("{ADMIN_PREFIX}/v3/scanner/usage-state/reset").as_str(), AdminOperation(&ScannerUsageStateResetHandler {}), )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/scanner/usage-state/recovery-intents/{{intent_id}}").as_str(), + AdminOperation(&ScannerUsageStateRecoveryIntentStatusHandler {}), + )?; r.insert( Method::GET, format!("{ADMIN_PREFIX}/v3/ilm/expiry/status").as_str(), @@ -269,11 +294,91 @@ async fn validate_scanner_reset_request(req: &S3Request) -> S3Result) -> S3Result> { + json_response_with_status(StatusCode::OK, body) +} + +fn json_response_with_status(status: StatusCode, body: Vec) -> S3Result> { let mut headers = HeaderMap::new(); let content_type = HeaderValue::from_str(JSON_CONTENT_TYPE) .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid content type: {err}")))?; headers.insert(CONTENT_TYPE, content_type); - Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), headers)) + Ok(S3Response::with_headers((status, Body::from(body)), headers)) +} + +fn scanner_recovery_intent_error(err: rustfs_scanner::ScannerError) -> S3Error { + let message = err.to_string(); + let code = if message.contains("action is unsupported") + || message.contains("mode is unsupported") + || message.contains("actor identity is invalid") + || message.contains("intent id is invalid") + || message.contains("idempotency key") + || message.contains("requires") + { + S3ErrorCode::InvalidRequest + } else { + S3ErrorCode::InternalError + }; + S3Error::with_message(code, message) +} + +fn scanner_recovery_intent_record_response( + status_code: StatusCode, + status: &'static str, + record: rustfs_scanner::ScannerRecoveryIntentRecord, +) -> S3Result> { + let response = ScannerRecoveryIntentResponse { + status, + action: record.action, + mode: record.mode, + intent_id: record.intent_id, + state: record.state, + }; + let body = serde_json::to_vec(&response).map_err(|err| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("failed to encode scanner recovery intent response: {err}"), + ) + })?; + json_response_with_status(status_code, body) +} + +fn scanner_recovery_intent_accept_response( + result: rustfs_scanner::ScannerRecoveryIntentAcceptResult, +) -> S3Result> { + match result { + rustfs_scanner::ScannerRecoveryIntentAcceptResult::Accepted { record } => { + scanner_recovery_intent_record_response(StatusCode::ACCEPTED, "accepted", record) + } + rustfs_scanner::ScannerRecoveryIntentAcceptResult::Replayed { record } => { + scanner_recovery_intent_record_response(StatusCode::ACCEPTED, "replayed", record) + } + rustfs_scanner::ScannerRecoveryIntentAcceptResult::Conflict { existing } => { + let response = ScannerRecoveryIntentConflictResponse { + status: "conflict", + intent_id: existing.intent_id, + state: existing.state, + }; + let body = serde_json::to_vec(&response).map_err(|err| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("failed to encode scanner recovery intent conflict: {err}"), + ) + })?; + json_response_with_status(StatusCode::CONFLICT, body) + } + } +} + +fn scanner_recovery_intent_executor_id(result: &rustfs_scanner::ScannerRecoveryIntentAcceptResult) -> Option { + match result { + rustfs_scanner::ScannerRecoveryIntentAcceptResult::Accepted { record } + | rustfs_scanner::ScannerRecoveryIntentAcceptResult::Replayed { record } + if matches!(record.state.as_str(), "accepted" | "running") => + { + Some(record.intent_id.clone()) + } + _ => None, + } } pub struct ScannerStatusHandler {} @@ -314,6 +419,8 @@ pub struct ScannerCycleStateResetHandler {} pub struct ScannerUsageStateResetHandler {} +pub struct ScannerUsageStateRecoveryIntentStatusHandler {} + #[async_trait::async_trait] impl Operation for ScannerCycleStateResetHandler { async fn call(&self, mut req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -361,6 +468,35 @@ impl Operation for ScannerUsageStateResetHandler { .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"))?; + if reset.async_intent { + let idempotency_key = reset + .idempotency_key + .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "async reset requires idempotency_key"))?; + let request = rustfs_scanner::ScannerRecoveryIntentRequest { + action: rustfs_scanner::SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD.to_string(), + mode: reset.mode, + idempotency_key, + actor_sha256: rustfs_scanner::scanner_recovery_actor_sha256(&_cred.access_key), + }; + let executor_store = store.clone(); + let accepted = rustfs_scanner::accept_scanner_usage_recovery_intent(store, request) + .await + .map_err(scanner_recovery_intent_error)?; + if let Some(intent_id) = scanner_recovery_intent_executor_id(&accepted) { + tokio::spawn(async move { + let _ = rustfs_scanner::scanner::run_scanner_usage_recovery_intent( + CancellationToken::new(), + executor_store, + intent_id, + ) + .await; + }); + } + return scanner_recovery_intent_accept_response(accepted); + } + if reset.idempotency_key.is_some() { + return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "idempotency_key requires async reset")); + } 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 @@ -377,6 +513,27 @@ impl Operation for ScannerUsageStateResetHandler { } } +#[async_trait::async_trait] +impl Operation for ScannerUsageStateRecoveryIntentStatusHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let _cred = validate_scanner_reset_request(&req).await?; + let intent_id = params.get("intent_id").unwrap_or(""); + 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 record = rustfs_scanner::get_scanner_usage_recovery_intent(store, intent_id) + .await + .map_err(scanner_recovery_intent_error)? + .ok_or_else(|| { + let mut err = S3Error::with_message(S3ErrorCode::NoSuchKey, "scanner recovery intent not found"); + err.set_status_code(StatusCode::NOT_FOUND); + err + })?; + scanner_recovery_intent_record_response(StatusCode::OK, "found", record) + } +} + #[async_trait::async_trait] impl Operation for IlmExpiryStatusHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -472,6 +629,8 @@ mod tests { 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"); + assert!(!full_rebuild.async_intent); + assert!(full_rebuild.idempotency_key.is_none()); 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"); @@ -481,6 +640,110 @@ mod tests { ); } + #[test] + fn admin_usage_reset_accepts_explicit_async_intent_contract() { + let request: ScannerUsageStateResetRequest = + serde_json::from_str(r#"{"mode":"full-rebuild","async":true,"idempotency_key":"reset-key-0001"}"#) + .expect("async reset intent contract should parse"); + + assert_eq!(request.mode, "full-rebuild"); + assert!(request.async_intent); + assert_eq!(request.idempotency_key.as_deref(), Some("reset-key-0001")); + } + + #[test] + fn scanner_recovery_intent_response_uses_accepted_status() { + let record = rustfs_scanner::ScannerRecoveryIntentRecord { + schema_version: 1, + intent_id: "0".repeat(64), + action: rustfs_scanner::SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD.to_string(), + mode: "full-rebuild".to_string(), + state: "accepted".to_string(), + actor_sha256: "1".repeat(64), + idempotency_key_sha256: "2".repeat(64), + request_sha256: "3".repeat(64), + accepted_at_unix_secs: 7, + }; + + let response = + scanner_recovery_intent_accept_response(rustfs_scanner::ScannerRecoveryIntentAcceptResult::Accepted { record }) + .expect("accepted intent response"); + + assert_eq!(response.output.0, StatusCode::ACCEPTED); + } + + #[test] + fn scanner_recovery_intent_executor_only_starts_non_terminal_work() { + let mut record = rustfs_scanner::ScannerRecoveryIntentRecord { + schema_version: 1, + intent_id: "0".repeat(64), + action: rustfs_scanner::SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD.to_string(), + mode: "full-rebuild".to_string(), + state: "accepted".to_string(), + actor_sha256: "1".repeat(64), + idempotency_key_sha256: "2".repeat(64), + request_sha256: "3".repeat(64), + accepted_at_unix_secs: 7, + }; + + assert_eq!( + scanner_recovery_intent_executor_id(&rustfs_scanner::ScannerRecoveryIntentAcceptResult::Accepted { + record: record.clone(), + }) + .as_deref(), + Some(record.intent_id.as_str()) + ); + record.state = "running".to_string(); + assert_eq!( + scanner_recovery_intent_executor_id(&rustfs_scanner::ScannerRecoveryIntentAcceptResult::Replayed { + record: record.clone(), + }) + .as_deref(), + Some(record.intent_id.as_str()) + ); + record.state = "completed".to_string(); + assert!( + scanner_recovery_intent_executor_id(&rustfs_scanner::ScannerRecoveryIntentAcceptResult::Replayed { + record: record.clone(), + }) + .is_none() + ); + assert!( + scanner_recovery_intent_executor_id(&rustfs_scanner::ScannerRecoveryIntentAcceptResult::Conflict { + existing: rustfs_scanner::ScannerRecoveryIntentConflict { + intent_id: record.intent_id, + state: "accepted".to_string(), + }, + }) + .is_none() + ); + } + + #[test] + fn scanner_recovery_intent_response_reports_conflict_status() { + let response = scanner_recovery_intent_accept_response(rustfs_scanner::ScannerRecoveryIntentAcceptResult::Conflict { + existing: rustfs_scanner::ScannerRecoveryIntentConflict { + intent_id: "0".repeat(64), + state: "accepted".to_string(), + }, + }) + .expect("conflict intent response"); + + assert_eq!(response.output.0, StatusCode::CONFLICT); + } + + #[test] + fn scanner_recovery_intent_error_keeps_persisted_corruption_server_side() { + let invalid_actor = + scanner_recovery_intent_error(rustfs_scanner::ScannerError::Other("actor identity is invalid".to_string())); + assert_eq!(invalid_actor.code(), &S3ErrorCode::InvalidRequest); + + let corrupt_record = scanner_recovery_intent_error(rustfs_scanner::ScannerError::Other( + "scanner recovery intent is invalid: expected value".to_string(), + )); + assert_eq!(corrupt_record.code(), &S3ErrorCode::InternalError); + } + #[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 54d1715d6..aebf92849 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -489,6 +489,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ CONFIG_UPDATE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/scanner/usage-state/recovery-intents/{intent_id}", + CONFIG_UPDATE, + RouteRiskLevel::High, + ), admin( HttpMethod::Get, "/rustfs/admin/v3/ilm/expiry/status", @@ -2181,6 +2187,20 @@ mod tests { assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/usage-state/reset", SERVER_INFO); } + #[test] + fn route_policy_requires_config_update_for_scanner_usage_recovery_intent() { + assert_action( + HttpMethod::Get, + "/rustfs/admin/v3/scanner/usage-state/recovery-intents/{intent_id}", + CONFIG_UPDATE, + ); + assert_not_action( + HttpMethod::Get, + "/rustfs/admin/v3/scanner/usage-state/recovery-intents/{intent_id}", + SERVER_INFO, + ); + } + #[test] fn route_policy_uses_tier_actions_for_transition_routes() { assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/recovery/records", LIST_TIER); diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 8da3b43e2..5a1e66ced 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -290,6 +290,11 @@ fn expected_admin_route_matrix() -> Vec { 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_sample( + Method::GET, + "/v3/scanner/usage-state/recovery-intents/{intent_id}", + "/v3/scanner/usage-state/recovery-intents/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ), admin_route(Method::GET, "/v3/audit/target/list"), admin_route_sample( Method::PUT, @@ -945,6 +950,11 @@ fn test_register_routes_cover_representative_admin_paths() { 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/scanner/usage-state/recovery-intents/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + ); assert_route(&router, Method::GET, &admin_path("/v3/ilm/expiry/status")); assert_route(&router, Method::GET, &admin_path("/v3/ilm/recovery/records")); assert_route( @@ -1458,6 +1468,12 @@ fn test_admin_alias_paths_match_existing_admin_routes() { (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/scanner/usage-state/recovery-intents/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ), + ), (Method::GET, compat_admin_alias_path("/v3/ilm/expiry/status")), (Method::PUT, compat_admin_alias_path("/v3/on-demand-migration/b")), (Method::GET, compat_admin_alias_path("/v3/on-demand-migration/b")),