From 99c1f4418bf5990e3ec041a601a6622c50036c36 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 09:16:33 +0800 Subject: [PATCH] fix(scanner): expose recovery intent identity (#7445) Co-authored-by: zhi22915 --- crates/madmin/src/client.rs | 41 +++++++++++++++++++++++++++- rustfs/src/admin/handlers/scanner.rs | 26 ++++++++++++++---- 2 files changed, 60 insertions(+), 7 deletions(-) diff --git a/crates/madmin/src/client.rs b/crates/madmin/src/client.rs index f76c1048f..bd68be97e 100644 --- a/crates/madmin/src/client.rs +++ b/crates/madmin/src/client.rs @@ -263,6 +263,14 @@ pub struct ScannerUsageRecoveryIntentResponse { pub mode: String, pub intent_id: String, pub state: String, + #[serde(default)] + pub actor_sha256: Option, + #[serde(default)] + pub idempotency_key_sha256: Option, + #[serde(default)] + pub request_sha256: Option, + #[serde(default)] + pub accepted_at_unix_secs: Option, #[serde(flatten)] pub extra: serde_json::Map, } @@ -869,6 +877,10 @@ mod tests { "mode": "full-rebuild", "intent_id": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", "state": "accepted", + "actor_sha256": "1111111111111111111111111111111111111111111111111111111111111111", + "idempotency_key_sha256": "2222222222222222222222222222222222222222222222222222222222222222", + "request_sha256": "3333333333333333333333333333333333333333333333333333333333333333", + "accepted_at_unix_secs": 7, "future": {"worker": "pending"} })) .unwrap(); @@ -877,7 +889,33 @@ mod tests { assert_eq!(intent.mode, "full-rebuild"); assert_eq!(intent.intent_id, "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"); assert_eq!(intent.state, "accepted"); + assert_eq!( + intent.actor_sha256.as_deref(), + Some("1111111111111111111111111111111111111111111111111111111111111111") + ); + assert_eq!( + intent.idempotency_key_sha256.as_deref(), + Some("2222222222222222222222222222222222222222222222222222222222222222") + ); + assert_eq!( + intent.request_sha256.as_deref(), + Some("3333333333333333333333333333333333333333333333333333333333333333") + ); + assert_eq!(intent.accepted_at_unix_secs, Some(7)); assert_eq!(intent.extra["future"]["worker"], "pending"); + + let legacy_intent: ScannerUsageRecoveryIntentResponse = serde_json::from_value(json!({ + "status": "accepted", + "action": "usage-full-rebuild", + "mode": "full-rebuild", + "intent_id": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + "state": "accepted" + })) + .unwrap(); + assert!(legacy_intent.actor_sha256.is_none()); + assert!(legacy_intent.idempotency_key_sha256.is_none()); + assert!(legacy_intent.request_sha256.is_none()); + assert!(legacy_intent.accepted_at_unix_secs.is_none()); } #[test] @@ -971,7 +1009,7 @@ mod tests { #[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"}"#, + r#"{"status":"accepted","action":"usage-full-rebuild","mode":"full-rebuild","intent_id":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","state":"accepted","actor_sha256":"1111111111111111111111111111111111111111111111111111111111111111","idempotency_key_sha256":"2222222222222222222222222222222222222222222222222222222222222222","request_sha256":"3333333333333333333333333333333333333333333333333333333333333333","accepted_at_unix_secs":7}"#, 202, ) .await; @@ -985,6 +1023,7 @@ mod tests { assert_eq!(accepted.status, "accepted"); assert_eq!(accepted.mode, "full-rebuild"); assert_eq!(accepted.state, "accepted"); + assert_eq!(accepted.accepted_at_unix_secs, Some(7)); let request = server.recorded(); assert_eq!(request.method, "POST"); assert_eq!(request.path, "/rustfs/admin/v3/scanner/usage-state/reset"); diff --git a/rustfs/src/admin/handlers/scanner.rs b/rustfs/src/admin/handlers/scanner.rs index f96f71406..bb2e83d44 100644 --- a/rustfs/src/admin/handlers/scanner.rs +++ b/rustfs/src/admin/handlers/scanner.rs @@ -75,6 +75,10 @@ struct ScannerRecoveryIntentResponse { mode: String, intent_id: String, state: String, + actor_sha256: String, + idempotency_key_sha256: String, + request_sha256: String, + accepted_at_unix_secs: u64, } #[derive(Debug, Serialize)] @@ -332,6 +336,10 @@ fn scanner_recovery_intent_record_response( mode: record.mode, intent_id: record.intent_id, state: record.state, + actor_sha256: record.actor_sha256, + idempotency_key_sha256: record.idempotency_key_sha256, + request_sha256: record.request_sha256, + accepted_at_unix_secs: record.accepted_at_unix_secs, }; let body = serde_json::to_vec(&response).map_err(|err| { S3Error::with_message( @@ -371,10 +379,7 @@ fn scanner_recovery_intent_accept_response( 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") => - { + rustfs_scanner::ScannerRecoveryIntentAcceptResult::Accepted { record } if record.state == "accepted" => { Some(record.intent_id.clone()) } _ => None, @@ -673,7 +678,7 @@ mod tests { } #[test] - fn scanner_recovery_intent_executor_only_starts_non_terminal_work() { + fn scanner_recovery_intent_executor_starts_only_newly_accepted_work() { let mut record = rustfs_scanner::ScannerRecoveryIntentRecord { schema_version: 1, intent_id: "0".repeat(64), @@ -699,7 +704,16 @@ mod tests { record: record.clone(), }) .as_deref(), - Some(record.intent_id.as_str()) + None, + "a lost-response retry must not start a duplicate executor" + ); + record.state = "accepted".to_string(); + assert!( + scanner_recovery_intent_executor_id(&rustfs_scanner::ScannerRecoveryIntentAcceptResult::Replayed { + record: record.clone(), + }) + .is_none(), + "replayed accepted records remain durable for startup/control recovery instead of duplicating work" ); record.state = "completed".to_string(); assert!(