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)

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`; 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` |
+| `POST /kms/create-key`, `POST /kms/key/create` | `kms:Configure` | high | no | Legacy `mc` aliases of `POST /kms/keys` |
 | `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 |
 

docs/testing/ci-gates.md

@@ -228,6 +228,19 @@ these. The external `rustfs/auto-testing` functional workflows propagate suite
 failures. Their workflow status does not establish this registry's required
 case coverage, build provenance, or object-level oracles.
 
+For automation, `--check-scanner-heal-release "$RUN_DIR"` emits one compact
+JSON decision and exits nonzero while blocked. `verified_cases` contains only
+cases that pass the complete receipt, build provenance, nextest/JUnit and real
+oracle checks; `rejected_cases` names registered cases that do not, and
+`pending_gates` names the unimplemented release requirements. Approval requires
+every registered case to verify, `pending_gates` to be empty, and a future
+registry schema capable of representing the complete release matrix. Schema 1
+is deliberately marked `release_schema_capable: false`: it models only the
+single-version, unversioned-object restart/crash cases and cannot represent
+mixed-version, rollback, EC8+4 or performance evidence. A focused run,
+synthetic harness, compile-only result, skipped/retried test, ordinary CI
+success, or removal of pending text therefore cannot become a release approval.
+
 Run parser/receipt regressions with
 `scripts/python_bin.sh scripts/check_test_wiring.py --self-test`. Those fixtures
 validate the checker only and produce no runtime or performance evidence.

rustfs/src/admin/handlers/kms_keys.rs

@@ -18,7 +18,6 @@ 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;
@@ -198,25 +197,6 @@ 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
@@ -352,8 +332,9 @@ 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 = legacy_create_key_name(&req.uri, &tags)?;
+        let key_name = tags.get("name").cloned();
 
         let kms_request = CreateKeyRequest {
             key_name,
@@ -498,8 +479,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, legacy_create_key_name, parse_list_limit, scoped_key_id,
-        stable_json_value, wants_key_impact,
+        kms_generate_data_key_actions, kms_list_keys_actions, parse_list_limit, scoped_key_id, stable_json_value,
+        wants_key_impact,
     };
     use http::Uri;
     use hyper::StatusCode;
@@ -520,46 +501,6 @@ 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,23 +21,6 @@ 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
@@ -385,10 +368,9 @@ 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 let Some(status) = status {
-            s3e.set_status_code(status);
+        if matches!(s3e.code(), S3ErrorCode::Custom(code) if &**code == MAX_VERSIONS_EXCEEDED_CODE) {
+            s3e.set_status_code(StatusCode::BAD_REQUEST);
         }
         if let Some(source) = err.source {
             s3e.set_source(source);
@@ -460,19 +442,6 @@ 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 {
@@ -1011,25 +980,6 @@ 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,12 +4336,9 @@ 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]
@@ -5563,37 +5560,6 @@ 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.

scripts/check_test_wiring.py

@@ -1094,6 +1094,38 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li
         return [f"scanner/heal evidence rejected: {error}"]
 
 
+def scanner_heal_release_status(root: Path, directory: Path) -> dict[str, object]:
+    """Return a compact release decision without weakening case validation."""
+    registry = read_json(root / ".config/scanner-heal-required-tests.json")
+    evidence_integer(registry.get("schema"), "registry schema", 1, 1)
+    cases = registry.get("cases")
+    require(isinstance(cases, dict) and cases, "invalid scanner/heal registry")
+    pending = registry.get("release_pending")
+    require(isinstance(pending, dict), "invalid scanner/heal release requirements")
+    for gate, reason in pending.items():
+        require(isinstance(gate, str) and re.fullmatch(r"[A-Z][A-Z0-9-]*", gate) is not None,
+                "invalid scanner/heal release gate")
+        require(isinstance(reason, str) and reason.strip(), f"missing release requirement for {gate}")
+
+    verified_cases = []
+    rejected_cases = []
+    for case_id in sorted(cases):
+        if check_scanner_heal_evidence(root, directory, case_id):
+            rejected_cases.append(case_id)
+        else:
+            verified_cases.append(case_id)
+
+    return {
+        "schema": 1,
+        "decision": "blocked",
+        "release_approved": False,
+        "release_schema_capable": False,
+        "verified_cases": verified_cases,
+        "rejected_cases": rejected_cases,
+        "pending_gates": sorted(pending),
+    }
+
+
 def validate(root: Path) -> list[str]:
     errors: list[str] = []
     errors.extend(check_core_fixtures(root))
@@ -1311,6 +1343,61 @@ class SelfTests(unittest.TestCase):
             self.assertTrue(any(error.startswith("pending R-D:") for error in errors))
             self.assertTrue(any(error.startswith("pending R-L:") for error in errors))
 
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["rejected_cases"], [])
+            self.assertEqual(len(status["pending_gates"]), 21)
+
+    def test_scanner_heal_case_only_schema_cannot_approve_release(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            registry = read_json(root / ".config/scanner-heal-required-tests.json")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertFalse(status["release_schema_capable"])
+            self.assertEqual(status["rejected_cases"], [])
+            self.assertEqual(status["pending_gates"], [])
+
+    def test_scanner_heal_release_status_rejects_synthetic_case(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            registry = read_json(root / ".config/scanner-heal-required-tests.json")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+            path = run_dir / "background-target-crash.json"
+            oracle = read_json(path)
+            oracle["evidence"] = "synthetic"
+            write_json(path, oracle)
+            (run_dir / "execution.json").unlink()
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["rejected_cases"], ["background-target-crash"])
+            self.assertEqual(status["pending_gates"], [])
+
+    def test_scanner_heal_release_status_rejects_focused_case_run(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            registry = read_json(root / ".config/scanner-heal-required-tests.json")
+            registry["release_pending"] = {}
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+            (run_dir / "background-target-crash.json").unlink()
+            (run_dir / "execution.json").unlink()
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            status = scanner_heal_release_status(root, run_dir)
+            self.assertEqual(status["decision"], "blocked")
+            self.assertFalse(status["release_approved"])
+            self.assertEqual(status["verified_cases"], ["background-target-restart"])
+            self.assertEqual(status["rejected_cases"], ["background-target-crash"])
+
     def test_scanner_heal_finish_collects_oracles_from_registry(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root, run_dir = self.scanner_heal_fixture(Path(tmp))
@@ -2158,7 +2245,8 @@ def main() -> int:
     if sys.argv[1:] == ["--self-test"]:
         suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)
         return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
-    if sys.argv[1:2] in (["--begin-scanner-heal"], ["--finish-scanner-heal"], ["--check-scanner-heal"]):
+    if sys.argv[1:2] in (["--begin-scanner-heal"], ["--finish-scanner-heal"], ["--check-scanner-heal"],
+                         ["--check-scanner-heal-release"]):
         try:
             if len(sys.argv) == 5 and sys.argv[1] == "--begin-scanner-heal":
                 begin_scanner_heal_receipt(ROOT, Path(sys.argv[2]), Path(sys.argv[3]), Path(sys.argv[4]))
@@ -2173,7 +2261,16 @@ def main() -> int:
                 if not errors:
                     print(f"Case evidence verified: {sys.argv[3]}; this does not approve release")
                 return 1 if errors else 0
-            raise ValueError("expected --begin-scanner-heal DIR BINARY TEST_BINARY, --finish-scanner-heal DIR EXIT, or --check-scanner-heal DIR CASE|release")
+            if len(sys.argv) == 3 and sys.argv[1] == "--check-scanner-heal-release":
+                try:
+                    status = scanner_heal_release_status(ROOT, Path(sys.argv[2]))
+                except (OSError, KeyError, TypeError, ValueError, ET.ParseError) as error:
+                    print(json.dumps({"schema": 1, "decision": "invalid", "release_approved": False,
+                                      "error": str(error)}, sort_keys=True, separators=(",", ":")))
+                    return 2
+                print(json.dumps(status, sort_keys=True, separators=(",", ":")))
+                return 0 if status["release_approved"] else 1
+            raise ValueError("expected --begin-scanner-heal DIR BINARY TEST_BINARY, --finish-scanner-heal DIR EXIT, --check-scanner-heal DIR CASE|release, or --check-scanner-heal-release DIR")
         except (OSError, KeyError, TypeError, ValueError, subprocess.SubprocessError) as error:
             print(f"ERROR: {error}", file=sys.stderr)
             return 1

scripts/run_scanner_heal_evidence_case.sh

@@ -90,15 +90,22 @@ PY
 
 release_gate_must_remain_blocked() {
     local run_dir="$1"
-    local output="$run_dir/release-check.txt"
-    if "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal "$run_dir" release >"$output" 2>&1; then
+    local output="$run_dir/release-status.json"
+    if "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal-release "$run_dir" >"$output"; then
         echo "release gate unexpectedly approved a single Scanner/Heal evidence run" >&2
         return 1
     fi
-    if ! grep -Eq 'required test not selected:|pending [A-Z0-9-]+:' "$output"; then
-        echo "release gate did not explain why the Scanner/Heal release remains blocked" >&2
-        return 1
-    fi
+    "$PYTHON_BIN" - "$output" <<'PY'
+import json
+import pathlib
+import sys
+
+status = json.loads(pathlib.Path(sys.argv[1]).read_text())
+if status.get("decision") != "blocked" or status.get("release_approved") is not False:
+    raise SystemExit("release status did not record a blocked decision")
+if status.get("release_schema_capable") is not False:
+    raise SystemExit("case-only evidence schema unexpectedly became release-capable")
+PY
 }
 
 run_self_test() {
@@ -242,5 +249,5 @@ fi
 "$PYTHON_BIN" "$ROOT/scripts/check_test_wiring.py" --check-scanner-heal "$RUN_DIR" "$CASE_ID"
 release_gate_must_remain_blocked "$RUN_DIR"
 echo "Scanner/Heal evidence case verified: $CASE_ID"
-echo "Release gate remains blocked; details: $RUN_DIR/release-check.txt"
+echo "Release gate remains blocked; status: $RUN_DIR/release-status.json"
 echo "Evidence directory: $RUN_DIR"