fix(heal): cleanup consumed MRF replay journals

Do not retain Accepted or Merged replay intents as startup anchors after they have been handed to the heal manager. Only refused or still-pending replay records keep the journal on disk until a successor snapshot can persist them.

This keeps successor snapshots limited to the pending queue, which lets successful replay remove both authoritative and legacy journal paths and restores the crash-boundary tests around successor flush.

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

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
(cherry picked from commit d5b8f49c9d)

crates/heal/src/heal/mrf_queue.rs

@@ -516,7 +516,6 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
 
 struct MrfRuntime {
     queue: MrfQueue,
-    retained_replay_intents: Vec<MrfIntent>,
     config: MrfConsumerConfig,
     new_since_flush: usize,
     /// True while the in-memory pending set has changed since the last
@@ -536,7 +535,7 @@ impl MrfRuntime {
     fn snapshot(&self) -> (Vec<u8>, Vec<u8>) {
         let mut authoritative = Vec::new();
         let mut legacy = Vec::new();
-        for intent in self.retained_replay_intents.iter().chain(self.queue.intents()) {
+        for intent in self.queue.intents() {
             let scoped_identity =
                 !matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) && intent.scope.is_some();
             if !encode_intent(intent, &mut authoritative) {
@@ -674,11 +673,10 @@ pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
 struct ReplayOutcome {
     replayed: usize,
     journal_on_disk: bool,
-    retained_replay_intents: Vec<MrfIntent>,
 }
 
-fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize, retained_replay_depth: usize) -> bool {
-    rearm_incomplete || pending_depth > 0 || retained_replay_depth > 0
+fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize) -> bool {
+    rearm_incomplete || pending_depth > 0
 }
 
 /// Shared replay core: read + decode + re-arm, then drain what fits. The
@@ -700,7 +698,6 @@ async fn replay_into(
                 return ReplayOutcome {
                     replayed: 0,
                     journal_on_disk: false,
-                    retained_replay_intents: Vec::new(),
                 };
             }
         },
@@ -740,13 +737,10 @@ async fn replay_into(
 
     // Drain the replayed intents immediately; whatever the manager refuses
     // stays armed in `queue` for the consumer's retry loop.
-    let mut retained_replay_intents = Vec::new();
     if backoff_until.is_none() {
         while let Some(mut intent) = queue.pop_front() {
             match submit_mrf_heal_request(manager, &intent).await {
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
-                    retained_replay_intents.push(intent);
-                }
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts < MRF_MAX_ATTEMPTS {
@@ -775,7 +769,7 @@ async fn replay_into(
             }
         }
     }
-    let journal_on_disk = if replay_must_retain_journal(rearm_incomplete, queue.depth(), retained_replay_intents.len()) {
+    let journal_on_disk = if replay_must_retain_journal(rearm_incomplete, queue.depth()) {
         true
     } else {
         !delete_journals().await
@@ -783,7 +777,6 @@ async fn replay_into(
     ReplayOutcome {
         replayed,
         journal_on_disk,
-        retained_replay_intents,
     }
 }
 
@@ -793,7 +786,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     let config = MrfConsumerConfig::default();
     let mut runtime = MrfRuntime {
         queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
-        retained_replay_intents: Vec::new(),
         config: config.clone(),
         new_since_flush: 0,
         dirty: false,
@@ -805,7 +797,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     // on disk whenever any replayed intent still needs a successor snapshot.
     let replay = replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await;
     runtime.journal_on_disk = replay.journal_on_disk;
-    runtime.retained_replay_intents = replay.retained_replay_intents;
     // Anything still pending (e.g. the manager was full and backoff armed)
     // must be re-persisted by the next flush before replay can delete the
     // startup anchor.
@@ -823,7 +814,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                     // provably current AND idle (a dirty or pending state
                     // gets one last persist attempt, matching the shutdown
                     // retry the unconditional flush used to provide).
-                    if runtime.dirty || runtime.queue.depth() > 0 || !runtime.retained_replay_intents.is_empty() {
+                    if runtime.dirty || runtime.queue.depth() > 0 {
                         runtime.flush().await;
                     }
                     tracing::info!(
@@ -852,7 +843,6 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                 match tick_action(
                     runtime.dirty,
                     runtime.queue.depth(),
-                    runtime.retained_replay_intents.len(),
                     runtime.journal_on_disk,
                 ) {
                     TickAction::Flush => {
@@ -867,8 +857,8 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                         runtime.dispatch(manager.as_ref()).await;
                     }
                     TickAction::DeleteJournal => {
-                        // Only remove a stale journal after every replayed
-                        // intent has a durable successor proof.
+                        // All replayed intents have either been accepted,
+                        // merged, or replaced by a pending successor snapshot.
                         if delete_journals().await {
                             runtime.journal_on_disk = false;
                             gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
@@ -897,13 +887,11 @@ enum TickAction {
     Idle,
 }
 
-fn tick_action(dirty: bool, depth: usize, retained_replay_depth: usize, journal_on_disk: bool) -> TickAction {
+fn tick_action(dirty: bool, depth: usize, journal_on_disk: bool) -> TickAction {
     if dirty {
         TickAction::Flush
     } else if depth > 0 {
         TickAction::Retry
-    } else if retained_replay_depth > 0 {
-        TickAction::Idle
     } else if journal_on_disk {
         TickAction::DeleteJournal
     } else {
@@ -936,73 +924,34 @@ mod tests {
 
         // Dirty dominates: a changed pending set flushes even when idle
         // otherwise.
-        assert!(matches!(tick_action(true, 0, 0, false), Flush));
-        assert!(matches!(tick_action(true, 3, 0, true), Flush));
+        assert!(matches!(tick_action(true, 0, false), Flush));
+        assert!(matches!(tick_action(true, 3, true), Flush));
 
         // Clean backlog: no rewrite, but keep draining so an expired
         // admission backoff retries on time.
-        assert!(matches!(tick_action(false, 1, 0, false), Retry));
-        assert!(matches!(tick_action(false, 2, 0, true), Retry));
-
-        // Replayed records accepted by the manager are still restart anchors
-        // until a durable successor proof can tombstone them.
-        assert!(matches!(tick_action(false, 0, 1, true), Idle));
+        assert!(matches!(tick_action(false, 1, false), Retry));
+        assert!(matches!(tick_action(false, 2, true), Retry));
 
         // Quiescent with a stale journal file on disk: remove it.
-        assert!(matches!(tick_action(false, 0, 0, true), DeleteJournal));
+        assert!(matches!(tick_action(false, 0, true), DeleteJournal));
 
         // Fully quiescent: nothing to do.
-        assert!(matches!(tick_action(false, 0, 0, false), Idle));
+        assert!(matches!(tick_action(false, 0, false), Idle));
     }
 
     #[test]
     fn replay_cleanup_retains_journal_for_unarmed_or_refused_records() {
         assert!(
-            replay_must_retain_journal(true, 0, 0),
+            replay_must_retain_journal(true, 0),
             "a rejected replay record still needs its disk anchor"
         );
         assert!(
-            replay_must_retain_journal(false, 1, 0),
+            replay_must_retain_journal(false, 1),
             "a Full admission retry must keep the startup journal until the next snapshot"
         );
         assert!(
-            replay_must_retain_journal(false, 0, 1),
-            "an accepted replay record still needs a durable successor before cleanup"
-        );
-        assert!(
-            !replay_must_retain_journal(false, 0, 0),
-            "only a fully consumed replay snapshot with no retained anchors may be deleted"
-        );
-    }
-
-    #[test]
-    fn retained_replay_anchor_remains_in_successor_snapshot() {
-        let retained = intent("accepted-replay", "object", 0);
-        let mut runtime = MrfRuntime {
-            queue: MrfQueue::new(8, 8192),
-            retained_replay_intents: vec![retained.clone()],
-            config: MrfConsumerConfig::default(),
-            new_since_flush: 0,
-            dirty: false,
-            journal_on_disk: true,
-            backoff_until: None,
-        };
-        assert_eq!(
-            runtime.queue.try_push_typed(intent("new-pending", "object", 0)),
-            MrfQueuePushResult::Enqueued
-        );
-
-        let (authoritative, legacy) = runtime.snapshot();
-        let (decoded, truncated) = decode_journal(&authoritative);
-        let (legacy_decoded, legacy_truncated) = decode_journal(&legacy);
-
-        assert_eq!(truncated, 0);
-        assert_eq!(legacy_truncated, 0);
-        assert_eq!(decoded.len(), 2);
-        assert_eq!(legacy_decoded.len(), 2);
-        assert!(
-            decoded.iter().any(|intent| intent.bucket == retained.bucket),
-            "accepted replay anchor must remain crash-replayable"
+            !replay_must_retain_journal(false, 0),
+            "only a fully consumed replay snapshot may be deleted"
         );
     }
 

docs/operations/kms-admin-contract.md

@@ -39,7 +39,7 @@ The wire prefix is `/rustfs/admin/v3`. `GET /kms/status` and `GET /kms/service-s
 | `POST /kms/restore/dry-run` | `kms:Restore` | sensitive | no | Preflight; writes nothing |
 | `POST /kms/restore` | `kms:Restore` | high | no | Requires `confirm_backup_id` and `confirm_conflict_policy` |
 | `POST /kms/restore/abort` | `kms:Restore` | high | no | Requires `confirm_target_key_dir` |
-| `POST /kms/create-key`, `POST /kms/key/create` | `kms:Configure` | high | no | Legacy `mc` aliases of `POST /kms/keys` |
+| `POST /kms/create-key`, `POST /kms/key/create` | `kms:Configure` | high | no | Legacy `mc` aliases of `POST /kms/keys`; the key name comes from the `key-id` query parameter (`mc`'s form) or the `name` tag, and a request carrying both with different values is refused with `400` |
 | `GET /kms/describe-key`, `GET /kms/key/status` | `kms:DescribeKey` | sensitive | yes | Legacy aliases of `GET /kms/keys/{key_id}` |
 | `GET /kms/list-keys` | `kms:ListKeys` | sensitive | no | Legacy alias of `GET /kms/keys`; same listing contract |
 

rustfs/src/admin/handlers/kms_keys.rs

@@ -18,6 +18,7 @@ use super::kms_audit::{KmsAdminAudit, KmsAdminOperation};
 use crate::admin::auth::{validate_admin_request, validate_admin_request_with_kms_key};
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{current_kms_runtime_service_manager, current_or_init_kms_runtime_service_manager};
+use crate::admin::storage_api::s3;
 use crate::admin::utils::extract_query_params;
 use crate::auth::{check_key_valid, get_session_token};
 use crate::kms_deletion_gate::current_key_impact;
@@ -197,6 +198,25 @@ fn extract_key_id(uri: &hyper::Uri) -> Option<String> {
         .find_map(|name| query_params.get(name).filter(|value| !value.is_empty()).cloned())
 }
 
+/// Name of the key a legacy create request asks for.
+///
+/// `mc admin kms key create <name>` sends the name as the `key-id` query
+/// parameter with no body, while RustFS clients send it as the `name` tag.
+/// Both are honored. A request carrying both has to agree with itself:
+/// picking one silently would create a key under a name the caller never
+/// sees in its own request.
+fn legacy_create_key_name(uri: &hyper::Uri, tags: &HashMap<String, String>) -> S3Result<Option<String>> {
+    let query_name = extract_key_id(uri);
+    let tag_name = tags.get("name").cloned();
+    match (query_name, tag_name) {
+        (Some(query), Some(tag)) if query != tag => Err(s3::error(
+            s3::S3ErrorCode::InvalidRequest,
+            format!("key name in the query ({query}) and in tags.name ({tag}) differ"),
+        )),
+        (query, tag) => Ok(query.or(tag)),
+    }
+}
+
 /// The `key_id` of a KMS admin request body, read without committing to the
 /// strict schema of the endpoint: the authorization gate needs the target key
 /// before the body is parsed for execution, and a body that fails the strict
@@ -332,9 +352,8 @@ impl Operation for CreateKeyHandler {
             return Err(s3_error!(InternalError, "kms service is not initialized"));
         };
 
-        // Extract key name from tags if provided
         let tags = request.tags.unwrap_or_default();
-        let key_name = tags.get("name").cloned();
+        let key_name = legacy_create_key_name(&req.uri, &tags)?;
 
         let kms_request = CreateKeyRequest {
             key_name,
@@ -479,8 +498,8 @@ mod tests {
         DescribeKmsKeyResponse, GenerateDataKeyApiRequest, GenerateDataKeyApiResponse, ListKeysApiResponse, ListKmsKeysResponse,
         delete_key_error_status, delete_request_from_query, extract_key_id, extract_query_params, key_impact_if_requested,
         key_list_filters, kms_create_key_actions, kms_delete_key_actions, kms_describe_key_actions,
-        kms_generate_data_key_actions, kms_list_keys_actions, parse_list_limit, scoped_key_id, stable_json_value,
-        wants_key_impact,
+        kms_generate_data_key_actions, kms_list_keys_actions, legacy_create_key_name, parse_list_limit, scoped_key_id,
+        stable_json_value, wants_key_impact,
     };
     use http::Uri;
     use hyper::StatusCode;
@@ -501,6 +520,46 @@ mod tests {
         assert!(!actions.contains(&action), "expected action list not to contain {action:?}");
     }
 
+    #[test]
+    fn legacy_create_key_name_honors_the_minio_key_id_query() {
+        let uri: Uri = "/rustfs/admin/v3/kms/key/create?key-id=minio-key"
+            .parse()
+            .expect("uri should parse");
+
+        let name = legacy_create_key_name(&uri, &HashMap::new()).expect("a query-only name is valid");
+        assert_eq!(name.as_deref(), Some("minio-key"));
+    }
+
+    #[test]
+    fn legacy_create_key_name_falls_back_to_the_name_tag() {
+        let uri: Uri = "/rustfs/admin/v3/kms/key/create".parse().expect("uri should parse");
+        let tags = HashMap::from([("name".to_string(), "tagged-key".to_string())]);
+
+        let name = legacy_create_key_name(&uri, &tags).expect("a tag-only name is valid");
+        assert_eq!(name.as_deref(), Some("tagged-key"));
+        assert_eq!(legacy_create_key_name(&uri, &HashMap::new()).expect("no name is valid"), None);
+    }
+
+    #[test]
+    fn legacy_create_key_name_accepts_agreeing_sources_and_refuses_conflicting_ones() {
+        let uri: Uri = "/rustfs/admin/v3/kms/key/create?key-id=minio-key"
+            .parse()
+            .expect("uri should parse");
+
+        let agreeing = HashMap::from([("name".to_string(), "minio-key".to_string())]);
+        let name = legacy_create_key_name(&uri, &agreeing).expect("agreeing sources are valid");
+        assert_eq!(name.as_deref(), Some("minio-key"));
+
+        let conflicting = HashMap::from([("name".to_string(), "other-key".to_string())]);
+        let refused = legacy_create_key_name(&uri, &conflicting).expect_err("conflicting names must be refused");
+        assert_eq!(*refused.code(), super::s3::S3ErrorCode::InvalidRequest);
+        assert!(
+            refused
+                .message()
+                .is_some_and(|message| message.contains("minio-key") && message.contains("other-key"))
+        );
+    }
+
     #[test]
     fn test_extract_key_id_supports_minio_aliases() {
         for (uri, expected) in [

rustfs/src/error.rs

@@ -21,6 +21,23 @@ use s3s::{S3Error, S3ErrorCode};
 const MAX_VERSIONS_EXCEEDED_CODE: &str = "MaxVersionsExceeded";
 const MAX_VERSIONS_EXCEEDED_MESSAGE: &str = "You've exceeded the limit on the number of versions you can create on this object";
 
+/// S3 error code for a request that names a KMS key the KMS does not hold.
+pub const KMS_KEY_NOT_FOUND_ERROR_CODE: &str = "KMS.NotFoundException";
+
+/// HTTP status of the error codes s3s cannot derive on its own.
+///
+/// s3s answers `None` for every `Custom` code, which the response layer turns
+/// into a 500; a code that means "your request named something that does not
+/// exist" has to say so itself.
+fn custom_error_status(code: &S3ErrorCode) -> Option<StatusCode> {
+    match code {
+        S3ErrorCode::Custom(custom) if &**custom == KMS_KEY_NOT_FOUND_ERROR_CODE || &**custom == MAX_VERSIONS_EXCEEDED_CODE => {
+            Some(StatusCode::BAD_REQUEST)
+        }
+        _ => None,
+    }
+}
+
 /// Marks a request body that exceeded a presigned upload size capability.
 ///
 /// This marker must survive the body-reader and storage layers so the client
@@ -368,9 +385,10 @@ fn error_chain_s3s_body_stream_error(err: &(dyn std::error::Error + 'static)) ->
 
 impl From<ApiError> for S3Error {
     fn from(err: ApiError) -> Self {
+        let status = custom_error_status(&err.code);
         let mut s3e = S3Error::with_message(err.code, err.message);
-        if matches!(s3e.code(), S3ErrorCode::Custom(code) if &**code == MAX_VERSIONS_EXCEEDED_CODE) {
-            s3e.set_status_code(StatusCode::BAD_REQUEST);
+        if let Some(status) = status {
+            s3e.set_status_code(status);
         }
         if let Some(source) = err.source {
             s3e.set_source(source);
@@ -442,6 +460,19 @@ impl From<StorageError> for ApiError {
                     source: Some(Box::new(err)),
                 };
             }
+
+            // A request header or bucket default naming a key the KMS does not
+            // hold is the caller's mistake to correct, and S3 reports it as
+            // 400 `KMS.NotFoundException`. Left to the fallthrough it became a
+            // 500 whose generic message hid which key was missing.
+            if let Some(rustfs_kms::KmsError::KeyNotFound { key_id }) = inner.downcast_ref::<rustfs_kms::KmsError>() {
+                let message = format!("KMS key not found: {key_id}");
+                return ApiError {
+                    code: S3ErrorCode::Custom(KMS_KEY_NOT_FOUND_ERROR_CODE.into()),
+                    message,
+                    source: Some(Box::new(err)),
+                };
+            }
         }
 
         let code = match &err {
@@ -496,9 +527,9 @@ impl From<StorageError> for ApiError {
 
         let message = if matches!(&err, StorageError::QuotaExceeded { .. }) {
             err.to_string()
-        } else if matches!(&err, StorageError::MaxVersionsExceeded) {
-            ApiError::error_code_to_message(&code)
-        } else if code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)) {
+        } else if matches!(&err, StorageError::MaxVersionsExceeded)
+            || (code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)))
+        {
             ApiError::error_code_to_message(&code)
         } else if code == S3ErrorCode::InternalError {
             err.to_string()
@@ -980,6 +1011,25 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_kms_key_not_found_maps_to_bad_request_kms_not_found_exception() {
+        let api_error = ApiError::from(StorageError::other(rustfs_kms::KmsError::key_not_found("no-such-key")));
+
+        assert_eq!(api_error.code, S3ErrorCode::Custom(KMS_KEY_NOT_FOUND_ERROR_CODE.into()));
+        assert_eq!(api_error.message, "KMS key not found: no-such-key");
+
+        // s3s knows no status for a custom code; the conversion has to supply it.
+        let s3_error = S3Error::from(api_error);
+        assert_eq!(s3_error.status_code(), Some(StatusCode::BAD_REQUEST));
+    }
+
+    #[test]
+    fn test_generated_error_codes_keep_their_own_status() {
+        let s3_error = S3Error::from(ApiError::from(StorageError::other(rustfs_kms::KmsError::backend_error("down"))));
+
+        assert_eq!(s3_error.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
+    }
+
     #[test]
     fn test_unknown_authoritative_quota_usage_maps_to_retryable_error() {
         let api_error = ApiError::from(QuotaError::UsageUnavailable {

rustfs/src/storage/sse.rs

@@ -4336,9 +4336,12 @@ mod tests {
     fn kms_operation_errors_preserve_retryability_classification() {
         let unavailable = kms_operation_error(rustfs_kms::KmsError::backend_error("connection refused"));
         let corrupt = kms_operation_error(rustfs_kms::KmsError::cryptographic_error("decrypt", "authentication failed"));
+        let missing = kms_operation_error(rustfs_kms::KmsError::key_not_found("no-such-key"));
 
         assert_eq!(unavailable.code, S3ErrorCode::ServiceUnavailable);
         assert_eq!(corrupt.code, S3ErrorCode::InternalError);
+        assert_eq!(missing.code, S3ErrorCode::Custom(crate::error::KMS_KEY_NOT_FOUND_ERROR_CODE.into()));
+        assert_eq!(super::kms_data_plane_error_class(&missing), "key_not_found");
     }
 
     #[test]
@@ -5560,6 +5563,37 @@ mod tests {
         reset_sse_dek_provider();
     }
 
+    /// A write whose resolved key — from the request header or a bucket
+    /// default rule — is unknown to the KMS must come back as the client
+    /// error S3 uses for it, all the way from the backend lookup. Answering
+    /// 500 here made a bucket default pointing at a deleted or mistyped key
+    /// look like a server outage (rustfs/backlog#2330, KMS-312).
+    #[tokio::test]
+    async fn kms_provider_reports_an_unknown_key_as_kms_not_found() {
+        let _guard = lock_sse_test_state().await;
+        reset_sse_dek_provider();
+
+        let manager = configure_test_global_local_kms().await;
+        let provider = KmsSseDekProvider::new_with_service_manager(manager)
+            .await
+            .expect("kms provider should initialize from the configured test manager");
+
+        let context = super::build_object_encryption_context("bucket", "object", None);
+        let error = provider
+            .generate_sse_dek(&context, "no-such-key")
+            .await
+            .expect_err("the Local backend must refuse a key it does not hold");
+        assert_eq!(
+            error.code,
+            S3ErrorCode::Custom(crate::error::KMS_KEY_NOT_FOUND_ERROR_CODE.into()),
+            "got {error:?}"
+        );
+        assert!(error.message.contains("no-such-key"), "the missing key must be named: {error:?}");
+        assert_eq!(super::kms_data_plane_error_class(&error), "key_not_found");
+
+        reset_sse_dek_provider();
+    }
+
     /// Objects without a rewrappable envelope — plaintext, SSE-C, or a
     /// MinIO-sealed opaque data key — are reported NotApplicable without any
     /// provider call.