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 <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* 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 <qiuzgang@gmail.com>

* feat(scanner): execute recovery intents asynchronously

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/madmin/src/client.rs

@@ -254,6 +254,19 @@ pub struct ScannerUsageStateResetResponse {
     pub extra: serde_json::Map<String, serde_json::Value>,
 }
 
+/// 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<String, serde_json::Value>,
+}
+
 #[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<ScannerUsageRecoveryIntentResponse, AdminClientError> {
+        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<ScannerUsageRecoveryIntentResponse, AdminClientError> {
+        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<serde_json::Value, AdminClientError> {
@@ -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}"#;

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,

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,

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<String>,
 }
 
+#[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<String, ScannerError> {
+    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<ScannerRecoveryIntentRecord, ScannerError> {
+    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<ScannerRecoveryIntentRecord, ScannerError> {
+    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<impl ScannerObjectIO>,
+    path: &str,
+) -> Result<Option<ScannerRecoveryIntentRecord>, 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<impl ScannerObjectIO>,
+    path: &str,
+) -> Result<Option<(ScannerRecoveryIntentRecord, DataUsageCacheRevision)>, 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<impl ScannerObjectIO>,
+    intent_id: &str,
+) -> Result<Option<ScannerRecoveryIntentRecord>, 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<impl ScannerObjectIO>,
+    request: ScannerRecoveryIntentRequest,
+) -> Result<ScannerRecoveryIntentAcceptResult, ScannerError> {
+    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<ECStore>,
+    intent_id: &str,
+    expected_states: &[&str],
+    next_state: &str,
+) -> Result<Option<ScannerRecoveryIntentRecord>, 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<ECStore>,
+    intent_id: String,
+) -> Result<Option<ScannerUsageStateResetResult>, 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(),

crates/scanner/src/scanner/tests/recovery_control.rs

@@ -97,6 +97,15 @@ async fn run_disabled_startup(ctx: CancellationToken, store: Arc<ECStore>) {
     );
 }
 
+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<ECStore>) {
     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() {