ci: stop a PR run once Test and Lint has failed (#5530)

On run 30674613104 the e2e, ILM and sftp lanes had all failed while Test and
Lint and the rio-v2 variant kept running past 70 minutes. The run's verdict was
settled; the remaining lanes were spending sm-standard-4 time on a result
nobody could act on, and with one full PR run needing about seven of those
runners, that time comes out of other PRs' queue time.

Two mechanisms, both scoped to pull_request so main pushes, the merge queue and
the weekly schedule keep the full failure signal:

- test-and-lint-protocols: fail-fast on PRs, so one failing protocol leg stops
  its sibling. This is the only part that also covers fork PRs, since it needs
  no token.
- test-and-lint: on failure, cancel the run through the REST API.

Only test-and-lint may cancel. The lanes that are not required checks
(protocols, ILM, e2e, s3-tests) must never hold that power: a flake in one of
them would turn the required "Test and Lint" into `cancelled`, which blocks the
merge. A maintainer can merge today with sftp red, and that has to stay true.

The cancel step uses curl, not `gh`: every existing `gh` call in this repo runs
on ubuntu-latest, and the sm-standard-* images are custom and trimmed, so `gh`
is not known to exist there. Fork PRs are excluded by an explicit condition
rather than left to fail, since their GITHUB_TOKEN is forced read-only and
job-level permissions cannot raise it.

Job-level permissions must list contents: read alongside actions: write —
job-level permissions replace the workflow block instead of merging with it,
and dropping contents would break this job's checkout and the repo-token the
setup action passes to setup-protoc. Because that token can now cancel runs and
delete Actions caches, the checkout also sets persist-credentials: false so a
PR's own build.rs or proc-macro cannot read it back out of .git/config.

Refs: rustfs/backlog#1598, rustfs/backlog#1599

.github/workflows/ci-docs-only.yml

@@ -12,18 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# Companion to ci.yml for the required "Test and Lint" status check.
+# Companion to ci.yml for the required "Test and Lint" and "Quick Checks"
+# status checks.
 #
 # ci.yml skips docs-only pull requests via paths-ignore, but the branch
-# ruleset requires a check named "Test and Lint" — without this workflow a
-# docs-only PR would wait on that check forever. This workflow triggers on
-# exactly the paths ci.yml ignores and reports an instant success under the
-# same job name. Mixed PRs trigger both workflows and the real check still
-# gates: a required check with any failing run blocks the merge.
+# ruleset requires checks named "Test and Lint" and "Quick Checks" — without
+# this workflow a docs-only PR would wait on those checks forever. This
+# workflow triggers on exactly the paths ci.yml ignores and reports success
+# under the same job names. Mixed PRs trigger both workflows and the real
+# checks still gate: a required check with any failing run blocks the merge.
 # https://docs.github.com/en/repositories/configuring-branches-and-merges-in-your-repository/defining-the-mergeability-of-pull-requests/troubleshooting-required-status-checks#handling-skipped-but-required-checks
 #
 # Keep the paths list below in sync with the pull_request paths-ignore list
-# in ci.yml.
+# in ci.yml, and keep the quick-checks steps below byte-identical to the
+# quick-checks job in ci.yml (see the comment on that job).
 
 name: Continuous Integration (docs only)
 
@@ -52,9 +54,63 @@ permissions:
   contents: read
 
 jobs:
+  # Deliberately NOT a bare `echo`. Once "Quick Checks" becomes a required
+  # check, ci.yml gates every expensive job behind it, so a mixed PR reports
+  # two check runs with this name: the real one (45-51s) and this companion.
+  # GitHub has no written contract for how it picks between same-named
+  # required check runs ("latest wins" vs "any failure blocks"), so instead of
+  # relying on ordering we make both runs execute the same commands against
+  # the same merge ref — their conclusions are then necessarily identical and
+  # the choice does not matter. Keep these steps byte-identical to the
+  # quick-checks job in ci.yml (a guard script that asserts this, and the paths
+  # sync below, is tracked in rustfs/backlog#1603).
+  #
+  # For a genuinely docs-only PR this adds no strictness (no code changed, so
+  # fmt and the guards always pass) and costs ~50s of ubuntu-latest.
+  quick-checks:
+    name: Quick Checks
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+
+      - name: Install ripgrep
+        run: sudo apt-get update && sudo apt-get install -y ripgrep
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+        with:
+          components: rustfmt
+
+      - name: Check code formatting
+        run: cargo fmt --all --check
+
+      - name: Check unsafe code allowances
+        run: ./scripts/check_unsafe_code_allowances.sh
+
+      - name: Check layered dependencies
+        run: ./scripts/check_layer_dependencies.sh
+
+      - name: Check architecture migration rules
+        run: ./scripts/check_architecture_migration_rules.sh
+
+      - name: Check tokio io-uring feature guard
+        run: ./scripts/check_no_tokio_io_uring.sh
+
+      - name: Check extension schema boundaries
+        run: ./scripts/check_extension_schema_boundaries.sh
+
+      - name: Check body-cache whitelist guard
+        run: ./scripts/check_body_cache_whitelist.sh
+
+      - name: Check no planning docs committed
+        run: ./scripts/check_no_planning_docs.sh
+
   test-and-lint:
     name: Test and Lint
     runs-on: ubuntu-latest
+    timeout-minutes: 10
     steps:
       - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7

.github/workflows/ci.yml

@@ -140,13 +140,26 @@ jobs:
   test-and-lint:
     name: Test and Lint
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 90
+    # Both lines are required. Job-level `permissions` replaces the workflow
+    # block rather than merging with it, so declaring only `actions: write`
+    # would drop `contents: read` and break this job's checkout and the
+    # repo-token the setup action hands to setup-protoc.
+    permissions:
+      contents: read
+      actions: write
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
       - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          # This job's token can cancel runs and delete Actions caches. Checkout
+          # otherwise writes it into .git/config, where a PR's own build.rs or
+          # proc-macro could read it back out.
+          persist-credentials: false
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -277,6 +290,48 @@ jobs:
       - name: Run rebalance/decommission migration proofs
         run: ./scripts/check_migration_gate_count.sh
 
+      # Early stop. Once this job has failed the PR cannot merge, so the sibling
+      # lanes are burning runners on a result nobody can act on: on run
+      # 30674613104 three lanes had already failed while Test and Lint and the
+      # rio-v2 variant kept going past 70 minutes.
+      #
+      # Only this job may cancel. The lanes that are NOT required checks
+      # (protocols, ILM, e2e, s3-tests) must never hold that power: a flake in
+      # one of them would turn the required "Test and Lint" into `cancelled`,
+      # which blocks the merge. Today a maintainer can merge with sftp red, and
+      # that has to stay true.
+      #
+      # These steps run last so the `if: always()` artifact upload above still
+      # captures logs and diagnostics before the run goes away.
+      - name: Annotate early-stop reason
+        if: failure() && github.event_name == 'pull_request'
+        run: |
+          echo "## CI early-stop" >> "$GITHUB_STEP_SUMMARY"
+          echo "Job \`${GITHUB_JOB}\` (Test and Lint) failed; cancelling run ${GITHUB_RUN_ID} to free runners." >> "$GITHUB_STEP_SUMMARY"
+          echo "Sibling jobs showing **cancelled** were stopped by this job, not by their own failure." >> "$GITHUB_STEP_SUMMARY"
+
+      # curl rather than `gh`: every existing `gh` call in this repo runs on
+      # ubuntu-latest, and the sm-standard-* images are custom and trimmed (they
+      # ship no C toolchain, see the e2e job below), so `gh` is not known to
+      # exist here.
+      #
+      # Fork PRs are excluded explicitly instead of relying on the error path:
+      # their GITHUB_TOKEN is forced read-only and job-level permissions cannot
+      # raise it, so the call would always 403. Skipping keeps their logs clean.
+      - name: Cancel run on failure (same-repo PR only)
+        if: >-
+          failure() && github.event_name == 'pull_request'
+          && github.event.pull_request.head.repo.full_name == github.repository
+        continue-on-error: true
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          curl -fsS -X POST \
+            -H "Authorization: Bearer ${GH_TOKEN}" \
+            -H "Accept: application/vnd.github+json" \
+            -H "X-GitHub-Api-Version: 2022-11-28" \
+            "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/cancel" || true
+
   # Dedicated serial lane for the ILM / lifecycle integration tests. These tests
   # drive the object layer through process-global singletons (the GLOBAL_ENV
   # ECStore, the global tier-config manager, background-expiry workers) and bind
@@ -290,6 +345,7 @@ jobs:
   test-ilm-integration-serial:
     name: ILM Integration (serial)
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 45
     env:
@@ -327,6 +383,7 @@ jobs:
   test-and-lint-rio-v2:
     name: Test and Lint (rio-v2)
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 60
     env:
@@ -354,10 +411,17 @@ jobs:
   test-and-lint-protocols:
     name: "Test and Lint (${{ matrix.features.name }})"
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 60
     strategy:
-      fail-fast: false
+      # On a PR, one failing protocol leg is enough to know the PR is not ready,
+      # so stop the sibling leg instead of paying another ~40 minutes for it.
+      # Everywhere else (main pushes, the merge queue, the weekly schedule) keep
+      # the full signal: there we want to know whether swift AND sftp are broken,
+      # not just whichever failed first. This is the only part of the early-stop
+      # work that also covers fork PRs, since it needs no token.
+      fail-fast: ${{ github.event_name == 'pull_request' }}
       matrix:
         features:
           - name: swift
@@ -389,6 +453,7 @@ jobs:
   build-rustfs-debug-binary:
     name: Build RustFS Debug Binary
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 30
     env:
@@ -419,6 +484,7 @@ jobs:
   build-rustfs-debug-binary-rio-v2:
     name: Build RustFS Debug Binary (rio-v2)
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 30
     env:
@@ -448,6 +514,14 @@ jobs:
 
   uring-integration:
     name: io_uring Integration (real)
+    # The pull_request trigger includes `closed` purely so the concurrency
+    # group cancels in-flight runs of a closed PR; every other job opts out of
+    # that run with this guard (or is skipped through its `needs` chain). This
+    # job had neither, so each closed/merged PR really ran the whole io_uring
+    # suite (measured 4m17s / 7m19s / 7m31s on runs 30678272341 / 30678117601 /
+    # 30662728539) and kept the cancellation run in progress for minutes.
+    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    needs: [ quick-checks ]
     # GitHub-hosted ubuntu-latest runs a recent kernel with io_uring and, unlike
     # a container, applies no seccomp filter that would block io_uring_setup — so
     # the probe succeeds and the tests exercise the real UringBackend/FdCache/
@@ -746,7 +820,13 @@ jobs:
   # evaluates ILM within ~2s of the due time, well inside the poll window.
   s3-lifecycle-behavior-tests:
     name: S3 Lifecycle Behavior Tests
-    needs: [ build-rustfs-debug-binary ]
+    # Also gated on e2e-tests, matching s3-implemented-tests: when the e2e smoke
+    # suite is already red this lane cannot tell us anything new, and it holds a
+    # sm-standard-4 for up to 30 minutes doing so. Both lanes only download the
+    # prebuilt debug binary (no cargo build), and s3-implemented-tests — which
+    # already waits on e2e-tests — finishes later anyway, so a green PR's total
+    # wall clock is unchanged.
+    needs: [ build-rustfs-debug-binary, e2e-tests ]
     runs-on: sm-standard-4
     timeout-minutes: 30
     steps:

Cargo.lock

@@ -9523,7 +9523,6 @@ dependencies = [
  "moka",
  "rand 0.10.2",
  "reqwest",
- "rustfs-s3-types",
  "rustfs-security-governance",
  "rustfs-utils",
  "rustify",

crates/e2e_test/src/object_lock/object_lock_test.rs

@@ -26,6 +26,7 @@
 
 use super::common::*;
 use aws_sdk_s3::Client;
+use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::{ByteStream, DateTimeFormat};
 use aws_sdk_s3::types::{
     CompletedMultipartUpload, CompletedPart, Delete, MetadataDirective, ObjectIdentifier, ObjectLockLegalHoldStatus,
@@ -2120,6 +2121,127 @@ async fn test_multipart_default_retention_fixed_at_create() {
 // Versioning Auto-Enable Tests
 // ============================================================================
 
+#[tokio::test]
+#[serial]
+async fn test_unretained_object_lock_object_delete_and_bucket_cleanup() {
+    init_logging();
+    info!("🧪 Test: Unretained Object Lock object delete and bucket cleanup (Issue #5339)");
+
+    let mut env = ObjectLockTestEnvironment::new()
+        .await
+        .expect("failed to create Object Lock test environment");
+    env.start_rustfs().await.expect("failed to start RustFS");
+
+    let bucket = "test-object-lock-delete-cleanup";
+    let key = "unretained-object";
+
+    env.create_object_lock_bucket(bucket)
+        .await
+        .expect("failed to create Object Lock bucket");
+    let client = env.s3_client();
+
+    let put_response = client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(ByteStream::from_static(b"unretained data"))
+        .send()
+        .await
+        .expect("failed to upload unretained object");
+    let object_version_id = put_response
+        .version_id()
+        .expect("Object Lock buckets must create versioned objects")
+        .to_string();
+
+    let delete_response = client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect("failed to create delete marker");
+    assert_eq!(delete_response.delete_marker(), Some(true));
+    let delete_marker_version_id = delete_response
+        .version_id()
+        .expect("Deleting without a version ID must create a delete marker")
+        .to_string();
+
+    let get_error = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect_err("GET must not return an object hidden by a delete marker");
+    assert_eq!(get_error.raw_response().map(|response| response.status().as_u16()), Some(404));
+    assert_eq!(get_error.as_service_error().and_then(|error| error.code()), Some("NoSuchKey"));
+
+    let listed_objects = client
+        .list_objects_v2()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list current objects");
+    assert!(
+        listed_objects.contents().iter().all(|object| object.key() != Some(key)),
+        "ListObjectsV2 must hide objects whose latest version is a delete marker"
+    );
+
+    let listed_versions = client
+        .list_object_versions()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list object versions");
+    assert!(
+        listed_versions
+            .versions()
+            .iter()
+            .any(|version| version.key() == Some(key) && version.version_id() == Some(object_version_id.as_str())),
+        "The data version must remain until it is explicitly deleted"
+    );
+    assert!(
+        listed_versions
+            .delete_markers()
+            .iter()
+            .any(|marker| marker.key() == Some(key) && marker.version_id() == Some(delete_marker_version_id.as_str())),
+        "ListObjectVersions must expose the delete marker"
+    );
+
+    client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(object_version_id)
+        .send()
+        .await
+        .expect("failed to delete the data version");
+    client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(delete_marker_version_id)
+        .send()
+        .await
+        .expect("failed to delete the delete marker");
+
+    let remaining_versions = client
+        .list_object_versions()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list versions after cleanup");
+    assert!(remaining_versions.versions().is_empty());
+    assert!(remaining_versions.delete_markers().is_empty());
+
+    client
+        .delete_bucket()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("Deleting every version must remove xl.meta so the bucket can be deleted normally");
+}
+
 #[tokio::test]
 #[serial]
 async fn test_versioning_auto_enabled_with_object_lock() {

crates/kms/Cargo.toml

@@ -64,10 +64,6 @@ md-5 = { workspace = true }
 arc-swap = { workspace = true }
 rustfs-utils = { workspace = true }
 rustfs-security-governance = { workspace = true }
-# `EventName` for KMS audit records. A leaf crate with no rustfs dependencies,
-# so the audit sink can live outside this crate without a second, drifting
-# copy of the event vocabulary.
-rustfs-s3-types = { workspace = true }
 
 # HTTP client for Vault
 reqwest = { workspace = true }

crates/kms/src/audit.rs

@@ -1,423 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Audit contract for KMS management operations.
-//!
-//! [`KmsManager`](crate::manager::KmsManager) builds a [`KmsAuditRecord`] for
-//! every management operation it serves and hands it to the installed
-//! [`KmsAuditSink`]. No sink is installed by default, so a deployment that
-//! does not consume KMS audit records pays nothing beyond the `Option` check.
-//!
-//! The record is deliberately a KMS-side type rather than an audit-crate
-//! entry: keeping the dependency edge out of this crate lets the server map
-//! records onto its existing audit pipeline (and its delivery semantics)
-//! without KMS having to know that pipeline exists.
-//!
-//! # Redaction
-//!
-//! A record has no field that can hold key material, and every value that
-//! originates from a caller passes through [`redact_encryption_context`]
-//! before it is stored. Adding a field here means re-checking that invariant.
-
-use crate::error::KmsError;
-use crate::types::OperationContext;
-use rustfs_s3_types::EventName;
-use sha2::{Digest, Sha256};
-use std::collections::{BTreeMap, HashMap};
-use std::time::Duration;
-use uuid::Uuid;
-
-/// Encryption-context keys that describe *where* an object lives rather than
-/// anything secret about it. Their values are recorded verbatim; every other
-/// key is reduced to a digest.
-const ENCRYPTION_CONTEXT_ALLOWLIST: [&str; 5] = ["bucket", "object", "object_key", "algorithm", "sse_type"];
-
-/// Prefix marking a value that was replaced by a digest of itself.
-const DIGEST_PREFIX: &str = "sha256:";
-
-/// Hex characters of the digest that are kept. Enough to correlate repeated
-/// values across records without being reversible in practice.
-const DIGEST_LEN: usize = 16;
-
-/// A KMS management operation that produces an audit record.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub enum KmsAuditOperation {
-    /// Creation of a new master key.
-    CreateKey,
-    /// Metadata lookup for a single key.
-    DescribeKey,
-    /// Enumeration of keys.
-    ListKeys,
-    /// Scheduling a key for deletion after the pending window.
-    ScheduleKeyDeletion,
-    /// Cancelling a previously scheduled deletion.
-    CancelKeyDeletion,
-    /// Returning a disabled key to service.
-    EnableKey,
-    /// Taking a key out of service without destroying it.
-    DisableKey,
-    /// Rotating a key to a new version.
-    RotateKey,
-    /// Irreversible removal of key material once the pending window expired.
-    DeleteKey,
-}
-
-impl KmsAuditOperation {
-    /// Stable operation name for audit consumers.
-    pub fn as_str(&self) -> &'static str {
-        match self {
-            KmsAuditOperation::CreateKey => "CreateKey",
-            KmsAuditOperation::DescribeKey => "DescribeKey",
-            KmsAuditOperation::ListKeys => "ListKeys",
-            KmsAuditOperation::ScheduleKeyDeletion => "ScheduleKeyDeletion",
-            KmsAuditOperation::CancelKeyDeletion => "CancelKeyDeletion",
-            KmsAuditOperation::EnableKey => "EnableKey",
-            KmsAuditOperation::DisableKey => "DisableKey",
-            KmsAuditOperation::RotateKey => "RotateKey",
-            KmsAuditOperation::DeleteKey => "DeleteKey",
-        }
-    }
-
-    /// Notification/audit event name carried by records for this operation.
-    pub fn event_name(&self) -> EventName {
-        match self {
-            KmsAuditOperation::CreateKey => EventName::KmsKeyCreated,
-            KmsAuditOperation::DescribeKey | KmsAuditOperation::ListKeys => EventName::KmsKeyAccessed,
-            KmsAuditOperation::ScheduleKeyDeletion => EventName::KmsKeyDeletionScheduled,
-            KmsAuditOperation::CancelKeyDeletion => EventName::KmsKeyDeletionCancelled,
-            KmsAuditOperation::EnableKey => EventName::KmsKeyEnabled,
-            KmsAuditOperation::DisableKey => EventName::KmsKeyDisabled,
-            KmsAuditOperation::RotateKey => EventName::KmsKeyRotated,
-            KmsAuditOperation::DeleteKey => EventName::KmsKeyDeleted,
-        }
-    }
-}
-
-/// Whether the audited operation completed.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub enum KmsAuditOutcome {
-    /// The operation completed and its effect is durable.
-    Success,
-    /// The operation failed; `error_class` carries the reason category.
-    Failure,
-}
-
-impl KmsAuditOutcome {
-    /// Stable outcome name for audit consumers.
-    pub fn as_str(&self) -> &'static str {
-        match self {
-            KmsAuditOutcome::Success => "success",
-            KmsAuditOutcome::Failure => "failure",
-        }
-    }
-}
-
-/// Coarse, stable classification of a failed KMS operation.
-///
-/// Audit consumers alert on these, so the strings are a wire contract: add
-/// new classes rather than renaming existing ones.
-pub fn error_class(error: &KmsError) -> &'static str {
-    match error {
-        KmsError::ConfigurationError { .. } => "configuration",
-        KmsError::KeyNotFound { .. } => "key_not_found",
-        KmsError::InvalidKey { .. } => "invalid_key",
-        KmsError::CryptographicError { .. } => "cryptographic",
-        KmsError::BackendError { .. } => "backend",
-        KmsError::AccessDenied { .. } => "access_denied",
-        KmsError::KeyAlreadyExists { .. } => "key_already_exists",
-        KmsError::InvalidOperation { .. } => "invalid_operation",
-        KmsError::InternalError { .. } => "internal",
-        KmsError::SerializationError { .. } => "serialization",
-        KmsError::IoError { .. } => "io",
-        KmsError::CacheError { .. } => "cache",
-        KmsError::ValidationError { .. } => "validation",
-        KmsError::UnsupportedAlgorithm { .. } => "unsupported_algorithm",
-        KmsError::InvalidKeySize { .. } => "invalid_key_size",
-        KmsError::ContextMismatch { .. } => "context_mismatch",
-        KmsError::OperationTimedOut { .. } => "timeout",
-        KmsError::OperationCancelled { .. } => "cancelled",
-        KmsError::MaterialMissing { .. } => "material_missing",
-        KmsError::MaterialCorrupt { .. } => "material_corrupt",
-        KmsError::MaterialAuthenticationFailed { .. } => "material_authentication_failed",
-        KmsError::UnsupportedFormatVersion { .. } => "unsupported_format_version",
-        KmsError::KeyVersionNotFound { .. } => "key_version_not_found",
-        KmsError::Backup(_) => "backup",
-        KmsError::UnsupportedCapability { .. } => "unsupported_capability",
-        KmsError::CredentialsUnavailable { .. } => "credentials_unavailable",
-    }
-}
-
-/// One audited KMS management operation.
-///
-/// Everything an audit consumer needs to answer "who did what to which key,
-/// against which backend, and did it work" — and nothing that could carry key
-/// material. See the module docs for the redaction rules.
-///
-/// Tenant attribution is intentionally absent: multi-tenancy is not modelled
-/// yet, and an invented tenant value is worse than a missing one. It will
-/// arrive as an entry in [`Self::context`] once tenancy lands.
-#[derive(Debug, Clone)]
-pub struct KmsAuditRecord {
-    /// Correlates every record emitted for one logical request.
-    pub operation_id: Uuid,
-    /// The audited operation.
-    pub operation: KmsAuditOperation,
-    /// Event name published to audit consumers.
-    pub event: EventName,
-    /// Authenticated identity that requested the operation, or
-    /// [`OperationContext::INTERNAL_PRINCIPAL`] for server-initiated work.
-    pub principal: String,
-    /// Client address, when the caller supplied one.
-    pub source_ip: Option<String>,
-    /// Client user agent, when the caller supplied one.
-    pub user_agent: Option<String>,
-    /// Key the operation acted on. `None` for operations that span keys, such
-    /// as listing.
-    pub key_id: Option<String>,
-    /// Key version the operation resolved to, when the result identifies one.
-    /// Operations on a key as a whole leave this unset.
-    pub key_version: Option<u32>,
-    /// Whether the operation succeeded.
-    pub outcome: KmsAuditOutcome,
-    /// Failure category; `None` on success. See [`error_class`].
-    pub error_class: Option<&'static str>,
-    /// Backend that served the operation, e.g. `local` or `vault-transit`.
-    pub backend: &'static str,
-    /// Wall-clock time spent in the audited operation.
-    pub latency: Duration,
-    /// Retries performed above the audit point. `None` means the number is
-    /// not observable here: backend-internal retries are accounted for by the
-    /// `rustfs_kms_backend_operation_attempts` metric instead of being
-    /// duplicated (and possibly contradicted) in the audit trail.
-    pub retry_count: Option<u32>,
-    /// Server-supplied correlation values carried by the operation context.
-    /// Also the reserved slot for tenant attribution.
-    pub context: BTreeMap<String, String>,
-    /// Caller-supplied encryption context, after [`redact_encryption_context`].
-    pub encryption_context: BTreeMap<String, String>,
-}
-
-impl KmsAuditRecord {
-    /// Start a record for `operation` performed under `context`.
-    ///
-    /// The outcome defaults to failure so that a record which somehow escapes
-    /// without [`Self::with_result`] under-reports success rather than
-    /// inventing it.
-    pub fn new(operation: KmsAuditOperation, context: &OperationContext, backend: &'static str) -> Self {
-        Self {
-            operation_id: context.operation_id,
-            operation,
-            event: operation.event_name(),
-            principal: context.principal.clone(),
-            source_ip: context.source_ip.clone(),
-            user_agent: context.user_agent.clone(),
-            key_id: None,
-            key_version: None,
-            outcome: KmsAuditOutcome::Failure,
-            error_class: None,
-            backend,
-            latency: Duration::ZERO,
-            retry_count: None,
-            context: context
-                .additional_context
-                .iter()
-                .map(|(k, v)| (k.clone(), v.clone()))
-                .collect(),
-            encryption_context: BTreeMap::new(),
-        }
-    }
-
-    /// Attach the key the operation acted on.
-    pub fn with_key_id(mut self, key_id: Option<impl Into<String>>) -> Self {
-        self.key_id = key_id.map(Into::into);
-        self
-    }
-
-    /// Attach the key version the operation resolved to.
-    pub fn with_key_version(mut self, key_version: Option<u32>) -> Self {
-        self.key_version = key_version;
-        self
-    }
-
-    /// Attach the measured duration of the operation.
-    pub fn with_latency(mut self, latency: Duration) -> Self {
-        self.latency = latency;
-        self
-    }
-
-    /// Derive outcome and error class from the operation result.
-    pub fn with_result<T>(mut self, result: &crate::error::Result<T>) -> Self {
-        match result {
-            Ok(_) => {
-                self.outcome = KmsAuditOutcome::Success;
-                self.error_class = None;
-            }
-            Err(error) => {
-                self.outcome = KmsAuditOutcome::Failure;
-                self.error_class = Some(error_class(error));
-            }
-        }
-        self
-    }
-
-    /// Attach the caller-supplied encryption context, redacted.
-    ///
-    /// Redaction happens here rather than at the call site so no caller can
-    /// place a raw context value into a record by accident.
-    pub fn with_encryption_context(mut self, encryption_context: &HashMap<String, String>) -> Self {
-        self.encryption_context = redact_encryption_context(encryption_context);
-        self
-    }
-}
-
-/// Reduce a caller-supplied encryption context to something safe to persist.
-///
-/// Keys naming the object's location are kept verbatim because they are the
-/// reason the context is audited at all. Every other value is replaced by a
-/// truncated digest, which still correlates repeated values across records
-/// but does not reproduce whatever the caller put there.
-pub fn redact_encryption_context(encryption_context: &HashMap<String, String>) -> BTreeMap<String, String> {
-    encryption_context
-        .iter()
-        .map(|(key, value)| {
-            let recorded = if ENCRYPTION_CONTEXT_ALLOWLIST.contains(&key.as_str()) {
-                value.clone()
-            } else {
-                digest_value(value)
-            };
-            (key.clone(), recorded)
-        })
-        .collect()
-}
-
-fn digest_value(value: &str) -> String {
-    let digest = hex::encode(Sha256::digest(value.as_bytes()));
-    format!("{DIGEST_PREFIX}{}", &digest[..DIGEST_LEN])
-}
-
-/// Receives KMS audit records.
-///
-/// Implementations must not block: they are called on the task that served
-/// the KMS operation. Delivery failures are the sink's problem — the KMS
-/// operation has already completed by the time a record is emitted, and its
-/// result is never changed by what the sink does.
-pub trait KmsAuditSink: Send + Sync {
-    /// Handle one audit record.
-    fn emit(&self, record: KmsAuditRecord);
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn every_operation_maps_to_a_kms_event() {
-        let operations = [
-            KmsAuditOperation::CreateKey,
-            KmsAuditOperation::DescribeKey,
-            KmsAuditOperation::ListKeys,
-            KmsAuditOperation::ScheduleKeyDeletion,
-            KmsAuditOperation::CancelKeyDeletion,
-            KmsAuditOperation::EnableKey,
-            KmsAuditOperation::DisableKey,
-            KmsAuditOperation::RotateKey,
-            KmsAuditOperation::DeleteKey,
-        ];
-
-        for operation in operations {
-            let event = operation.event_name();
-            assert!(event.is_kms(), "{} must map to a KMS event, got {event}", operation.as_str());
-        }
-    }
-
-    #[test]
-    fn record_defaults_to_failure_until_a_result_is_attached() {
-        let context = OperationContext::new("tester".to_string());
-        let record = KmsAuditRecord::new(KmsAuditOperation::CreateKey, &context, "local");
-
-        assert_eq!(record.outcome, KmsAuditOutcome::Failure);
-        assert!(record.error_class.is_none());
-
-        let ok: crate::error::Result<()> = Ok(());
-        assert_eq!(record.clone().with_result(&ok).outcome, KmsAuditOutcome::Success);
-
-        let denied: crate::error::Result<()> = Err(KmsError::access_denied("nope"));
-        let failed = record.with_result(&denied);
-        assert_eq!(failed.outcome, KmsAuditOutcome::Failure);
-        assert_eq!(failed.error_class, Some("access_denied"));
-    }
-
-    #[test]
-    fn record_carries_the_full_operation_context() {
-        let context = OperationContext::new("arn:aws:iam::user/alice".to_string())
-            .with_source_ip("10.0.0.7".to_string())
-            .with_user_agent("aws-cli/2".to_string())
-            .with_context("requestID".to_string(), "req-1".to_string());
-
-        let record = KmsAuditRecord::new(KmsAuditOperation::RotateKey, &context, "vault-transit")
-            .with_key_id(Some("key-1"))
-            .with_key_version(Some(3))
-            .with_latency(Duration::from_millis(12));
-
-        assert_eq!(record.operation_id, context.operation_id);
-        assert_eq!(record.principal, "arn:aws:iam::user/alice");
-        assert_eq!(record.source_ip.as_deref(), Some("10.0.0.7"));
-        assert_eq!(record.user_agent.as_deref(), Some("aws-cli/2"));
-        assert_eq!(record.key_id.as_deref(), Some("key-1"));
-        assert_eq!(record.key_version, Some(3));
-        assert_eq!(record.backend, "vault-transit");
-        assert_eq!(record.latency, Duration::from_millis(12));
-        assert_eq!(record.event, EventName::KmsKeyRotated);
-        assert_eq!(record.context.get("requestID").map(String::as_str), Some("req-1"));
-    }
-
-    #[test]
-    fn encryption_context_keeps_only_allowlisted_values_verbatim() {
-        let secret = "eyJhbGciOiJIUzI1NiJ9.super-secret-grant-token";
-        let context = HashMap::from([
-            ("bucket".to_string(), "photos".to_string()),
-            ("object_key".to_string(), "2026/cat.png".to_string()),
-            ("algorithm".to_string(), "AES256".to_string()),
-            ("grant_token".to_string(), secret.to_string()),
-            ("customer_reference".to_string(), "acct-4711".to_string()),
-        ]);
-
-        let redacted = redact_encryption_context(&context);
-
-        assert_eq!(redacted.get("bucket").map(String::as_str), Some("photos"));
-        assert_eq!(redacted.get("object_key").map(String::as_str), Some("2026/cat.png"));
-        assert_eq!(redacted.get("algorithm").map(String::as_str), Some("AES256"));
-
-        for key in ["grant_token", "customer_reference"] {
-            let value = redacted.get(key).expect("non-allowlisted key should be kept as a digest");
-            assert!(value.starts_with(DIGEST_PREFIX), "{key} should be digested, got {value}");
-        }
-
-        let rendered = format!("{redacted:?}");
-        assert!(!rendered.contains(secret), "redacted context must not reproduce the raw value");
-        assert!(!rendered.contains("acct-4711"), "redacted context must not reproduce the raw value");
-    }
-
-    #[test]
-    fn identical_values_digest_identically_across_records() {
-        // Correlation is the reason a digest is preferable to a fixed
-        // placeholder; assert it actually holds.
-        let first = redact_encryption_context(&HashMap::from([("tenant".to_string(), "alpha".to_string())]));
-        let second = redact_encryption_context(&HashMap::from([("tenant".to_string(), "alpha".to_string())]));
-        let other = redact_encryption_context(&HashMap::from([("tenant".to_string(), "beta".to_string())]));
-
-        assert_eq!(first.get("tenant"), second.get("tenant"));
-        assert_ne!(first.get("tenant"), other.get("tenant"));
-    }
-}

crates/kms/src/config.rs

@@ -130,21 +130,6 @@ pub enum KmsBackend {
     Static,
 }
 
-impl KmsBackend {
-    /// Stable identifier for logs, metrics, and audit records.
-    ///
-    /// External consumers key off these values, so treat them as a wire
-    /// contract rather than a rendering detail.
-    pub fn as_str(&self) -> &'static str {
-        match self {
-            KmsBackend::VaultKv2 => "vault-kv2",
-            KmsBackend::VaultTransit => "vault-transit",
-            KmsBackend::Local => "local",
-            KmsBackend::Static => "static",
-        }
-    }
-}
-
 /// Main KMS configuration
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct KmsConfig {

crates/kms/src/deletion_worker.rs

@@ -22,14 +22,12 @@
 //! every node of a deployment concurrently — a key is only ever removed while
 //! its (re-read) record is an expired pending deletion or a tombstone.
 
-use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink};
 use crate::backends::{ExpiredKeyRemoval, KmsBackend};
-use crate::error::Result;
-use crate::types::{KeyStatus, ListKeysRequest, OperationContext};
+use crate::types::{KeyStatus, ListKeysRequest};
 use async_trait::async_trait;
 use jiff::Zoned;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, warn};
 
@@ -70,8 +68,6 @@ pub(crate) struct DeletionWorker {
     default_key_id: Option<String>,
     reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
     interval: Duration,
-    backend_kind: &'static str,
-    audit_sink: Option<Arc<dyn KmsAuditSink>>,
 }
 
 impl DeletionWorker {
@@ -79,24 +75,15 @@ impl DeletionWorker {
         backend: Arc<dyn KmsBackend>,
         default_key_id: Option<String>,
         reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
-        backend_kind: &'static str,
     ) -> Self {
         Self {
             backend,
             default_key_id,
             reference_checker,
-            backend_kind,
-            audit_sink: None,
             interval: DEFAULT_SWEEP_INTERVAL,
         }
     }
 
-    /// Audit each irreversible key removal to `sink`.
-    pub(crate) fn with_audit_sink(mut self, sink: Option<Arc<dyn KmsAuditSink>>) -> Self {
-        self.audit_sink = sink;
-        self
-    }
-
     pub(crate) fn spawn(self, cancel: CancellationToken) -> tokio::task::JoinHandle<()> {
         tokio::spawn(async move { self.run(cancel).await })
     }
@@ -181,43 +168,15 @@ impl DeletionWorker {
 
         // The backend re-checks state and deadline under its own write
         // synchronization, so a cancellation racing this sweep wins there.
-        let started = Instant::now();
-        let outcome = self.backend.remove_expired_key(key_id, now).await;
-        match &outcome {
-            Ok(ExpiredKeyRemoval::Removed) => {
-                report.removed.push(key_id.to_string());
-                self.audit_removal(key_id, started, &outcome);
-            }
+        match self.backend.remove_expired_key(key_id, now).await {
+            Ok(ExpiredKeyRemoval::Removed) => report.removed.push(key_id.to_string()),
             Ok(ExpiredKeyRemoval::StateChanged | ExpiredKeyRemoval::NotExpired) => report.skipped += 1,
             Err(error) => {
                 warn!(key_id, %error, "failed to remove expired KMS key; will retry next sweep");
                 report.failed += 1;
-                self.audit_removal(key_id, started, &outcome);
             }
         }
     }
-
-    /// Record an attempted destruction of key material.
-    ///
-    /// Only attempts that reached the backend are audited: a key the sweep
-    /// declined to touch (not yet due, still referenced, state changed under
-    /// us) was never at risk, and recording it as a deletion event would
-    /// drown the real ones.
-    fn audit_removal(&self, key_id: &str, started: Instant, outcome: &Result<ExpiredKeyRemoval>) {
-        let Some(sink) = self.audit_sink.as_ref() else {
-            return;
-        };
-
-        // Removal runs on a background sweep, so there is no request
-        // principal to attribute it to.
-        let context = OperationContext::internal();
-        sink.emit(
-            KmsAuditRecord::new(KmsAuditOperation::DeleteKey, &context, self.backend_kind)
-                .with_key_id(Some(key_id))
-                .with_latency(started.elapsed())
-                .with_result(outcome),
-        );
-    }
 }
 
 #[cfg(test)]
@@ -257,7 +216,7 @@ mod tests {
     }
 
     fn worker(backend: Arc<LocalKmsBackend>) -> DeletionWorker {
-        DeletionWorker::new(backend, None, None, "local")
+        DeletionWorker::new(backend, None, None)
     }
 
     fn after_window() -> Zoned {
@@ -348,7 +307,7 @@ mod tests {
         schedule(&backend, &key_id).await;
 
         // Blocked while it is the configured default key.
-        let as_default = DeletionWorker::new(backend.clone(), Some(key_id.clone()), None, "local");
+        let as_default = DeletionWorker::new(backend.clone(), Some(key_id.clone()), None);
         let report = as_default.sweep(&after_window()).await;
         assert_eq!(report.blocked, vec![key_id.clone()]);
         assert!(report.removed.is_empty());
@@ -358,7 +317,6 @@ mod tests {
             backend.clone(),
             None,
             Some(Arc::new(StaticReferences(vec!["bucket:sse-bucket".to_string()]))),
-            "local",
         );
         let report = with_references.sweep(&after_window()).await;
         assert_eq!(report.blocked, vec![key_id.clone()]);
@@ -369,53 +327,11 @@ mod tests {
             .expect("blocked key must still exist");
 
         // Removed once nothing references it anymore.
-        let unreferenced = DeletionWorker::new(backend.clone(), None, Some(Arc::new(StaticReferences(Vec::new()))), "local");
+        let unreferenced = DeletionWorker::new(backend.clone(), None, Some(Arc::new(StaticReferences(Vec::new()))));
         let report = unreferenced.sweep(&after_window()).await;
         assert_eq!(report.removed, vec![key_id.clone()]);
     }
 
-    #[tokio::test]
-    async fn only_attempted_removals_are_audited() {
-        #[derive(Default)]
-        struct CapturingSink {
-            records: std::sync::Mutex<Vec<KmsAuditRecord>>,
-        }
-
-        impl KmsAuditSink for CapturingSink {
-            fn emit(&self, record: KmsAuditRecord) {
-                self.records.lock().expect("sink lock should not be poisoned").push(record);
-            }
-        }
-
-        let temp_dir = tempfile::tempdir().expect("temp dir");
-        let backend = local_backend(&temp_dir).await;
-        let key_id = create_key(&backend, "audited-removal").await;
-        schedule(&backend, &key_id).await;
-
-        let sink = Arc::new(CapturingSink::default());
-        let worker = DeletionWorker::new(backend.clone(), None, None, "local").with_audit_sink(Some(sink.clone()));
-
-        // A key that is not yet due was never at risk, so it produces no record.
-        worker.sweep(&Zoned::now()).await;
-        assert!(
-            sink.records.lock().expect("sink lock").is_empty(),
-            "a key the sweep declined to touch must not be audited as a deletion"
-        );
-
-        worker.sweep(&after_window()).await;
-
-        let records = sink.records.lock().expect("sink lock");
-        assert_eq!(records.len(), 1, "the removal should be audited exactly once");
-        let record = &records[0];
-        assert_eq!(record.operation, crate::audit::KmsAuditOperation::DeleteKey);
-        assert_eq!(record.event, rustfs_s3_types::EventName::KmsKeyDeleted);
-        assert_eq!(record.outcome, crate::audit::KmsAuditOutcome::Success);
-        assert_eq!(record.key_id.as_deref(), Some(key_id.as_str()));
-        assert_eq!(record.backend, "local");
-        // Background work has no request principal to attribute.
-        assert_eq!(record.principal, OperationContext::INTERNAL_PRINCIPAL);
-    }
-
     #[tokio::test]
     async fn deadline_survives_backend_restart_and_sweep_completes_it() {
         let temp_dir = tempfile::tempdir().expect("temp dir");

crates/kms/src/lib.rs

@@ -65,7 +65,6 @@
 
 // Core modules
 pub mod api_types;
-pub mod audit;
 pub mod backends;
 pub mod backup;
 mod cache;
@@ -87,7 +86,6 @@ pub use api_types::{
     StartKmsResponse, StopKmsResponse, TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse,
     UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
 };
-pub use audit::{KmsAuditOperation, KmsAuditOutcome, KmsAuditRecord, KmsAuditSink, redact_encryption_context};
 pub use config::*;
 pub use deletion_worker::DeletionReferenceChecker;
 pub use encryption::is_data_key_envelope;

crates/kms/src/manager.rs

@@ -14,7 +14,6 @@
 
 //! KMS manager for handling key operations and backend coordination
 
-use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink};
 use crate::backends::KmsBackend;
 use crate::cache::KmsCache;
 use crate::config::KmsConfig;
@@ -22,10 +21,9 @@ use crate::error::Result;
 use crate::types::{
     CancelKeyDeletionRequest, CancelKeyDeletionResponse, CreateKeyRequest, CreateKeyResponse, DecryptRequest, DecryptResponse,
     DeleteKeyRequest, DeleteKeyResponse, DescribeKeyRequest, DescribeKeyResponse, EncryptRequest, EncryptResponse,
-    GenerateDataKeyRequest, GenerateDataKeyResponse, ListKeysRequest, ListKeysResponse, OperationContext,
+    GenerateDataKeyRequest, GenerateDataKeyResponse, ListKeysRequest, ListKeysResponse,
 };
 use std::sync::Arc;
-use std::time::Instant;
 use tokio::sync::RwLock;
 
 /// KMS Manager coordinates operations between backends and caching
@@ -35,8 +33,6 @@ pub struct KmsManager {
     cache: Arc<RwLock<KmsCache>>,
     default_key_id: Option<String>,
     enable_cache: bool,
-    backend_kind: &'static str,
-    audit_sink: Option<Arc<dyn KmsAuditSink>>,
 }
 
 impl KmsManager {
@@ -48,72 +44,16 @@ impl KmsManager {
             cache,
             default_key_id: config.default_key_id,
             enable_cache: config.enable_cache,
-            backend_kind: config.backend.as_str(),
-            audit_sink: None,
         }
     }
 
-    /// Send an audit record for every management operation to `sink`.
-    ///
-    /// Without a sink the manager builds no records at all, so a deployment
-    /// that does not consume KMS audit records is unaffected.
-    pub fn with_audit_sink(mut self, sink: Arc<dyn KmsAuditSink>) -> Self {
-        self.audit_sink = Some(sink);
-        self
-    }
-
     /// Get the default key ID if configured
     pub fn get_default_key_id(&self) -> Option<&String> {
         self.default_key_id.as_ref()
     }
 
-    /// Emit an audit record for a completed management operation.
-    ///
-    /// Called after the operation resolved, so nothing here can change its
-    /// result; the record is built only when a sink is installed.
-    fn audit<T>(
-        &self,
-        operation: KmsAuditOperation,
-        context: &OperationContext,
-        key_id: Option<&str>,
-        started: Instant,
-        result: &Result<T>,
-    ) {
-        let Some(sink) = self.audit_sink.as_ref() else {
-            return;
-        };
-
-        sink.emit(
-            KmsAuditRecord::new(operation, context, self.backend_kind)
-                .with_key_id(key_id)
-                .with_latency(started.elapsed())
-                .with_result(result),
-        );
-    }
-
     /// Create a new master key
-    ///
-    /// Audited as an internal operation; callers serving an authenticated
-    /// request should use [`Self::create_key_with_context`].
     pub async fn create_key(&self, request: CreateKeyRequest) -> Result<CreateKeyResponse> {
-        self.create_key_with_context(request, &OperationContext::internal()).await
-    }
-
-    /// Create a new master key on behalf of `context`'s principal
-    pub async fn create_key_with_context(
-        &self,
-        request: CreateKeyRequest,
-        context: &OperationContext,
-    ) -> Result<CreateKeyResponse> {
-        let started = Instant::now();
-        let key_name = request.key_name.clone();
-        let result = self.create_key_inner(request).await;
-        let key_id = result.as_ref().ok().map(|r| r.key_id.as_str()).or(key_name.as_deref());
-        self.audit(KmsAuditOperation::CreateKey, context, key_id, started, &result);
-        result
-    }
-
-    async fn create_key_inner(&self, request: CreateKeyRequest) -> Result<CreateKeyResponse> {
         let response = self.backend.create_key(request).await?;
 
         // Cache the key metadata if enabled
@@ -144,27 +84,7 @@ impl KmsManager {
     }
 
     /// Describe a key
-    ///
-    /// Audited as an internal operation; callers serving an authenticated
-    /// request should use [`Self::describe_key_with_context`].
     pub async fn describe_key(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
-        self.describe_key_with_context(request, &OperationContext::internal()).await
-    }
-
-    /// Describe a key on behalf of `context`'s principal
-    pub async fn describe_key_with_context(
-        &self,
-        request: DescribeKeyRequest,
-        context: &OperationContext,
-    ) -> Result<DescribeKeyResponse> {
-        let started = Instant::now();
-        let key_id = request.key_id.clone();
-        let result = self.describe_key_inner(request).await;
-        self.audit(KmsAuditOperation::DescribeKey, context, Some(&key_id), started, &result);
-        result
-    }
-
-    async fn describe_key_inner(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
         // Check cache first if enabled
         if self.enable_cache {
             let cache = self.cache.read().await;
@@ -189,20 +109,8 @@ impl KmsManager {
     }
 
     /// List keys
-    ///
-    /// Audited as an internal operation; callers serving an authenticated
-    /// request should use [`Self::list_keys_with_context`].
     pub async fn list_keys(&self, request: ListKeysRequest) -> Result<ListKeysResponse> {
-        self.list_keys_with_context(request, &OperationContext::internal()).await
-    }
-
-    /// List keys on behalf of `context`'s principal
-    pub async fn list_keys_with_context(&self, request: ListKeysRequest, context: &OperationContext) -> Result<ListKeysResponse> {
-        let started = Instant::now();
-        let result = self.backend.list_keys(request).await;
-        // Listing spans keys, so the record carries no key id.
-        self.audit(KmsAuditOperation::ListKeys, context, None, started, &result);
-        result
+        self.backend.list_keys(request).await
     }
 
     /// Get cache statistics
@@ -225,27 +133,7 @@ impl KmsManager {
     }
 
     /// Delete a key
-    ///
-    /// Audited as an internal operation; callers serving an authenticated
-    /// request should use [`Self::delete_key_with_context`].
     pub async fn delete_key(&self, request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
-        self.delete_key_with_context(request, &OperationContext::internal()).await
-    }
-
-    /// Delete a key on behalf of `context`'s principal
-    pub async fn delete_key_with_context(
-        &self,
-        request: DeleteKeyRequest,
-        context: &OperationContext,
-    ) -> Result<DeleteKeyResponse> {
-        let started = Instant::now();
-        let key_id = request.key_id.clone();
-        let result = self.delete_key_inner(request).await;
-        self.audit(KmsAuditOperation::ScheduleKeyDeletion, context, Some(&key_id), started, &result);
-        result
-    }
-
-    async fn delete_key_inner(&self, request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
         let response = self.backend.delete_key(request).await?;
 
         // Remove from cache if enabled and key is being deleted
@@ -258,28 +146,7 @@ impl KmsManager {
     }
 
     /// Cancel key deletion
-    ///
-    /// Audited as an internal operation; callers serving an authenticated
-    /// request should use [`Self::cancel_key_deletion_with_context`].
     pub async fn cancel_key_deletion(&self, request: CancelKeyDeletionRequest) -> Result<CancelKeyDeletionResponse> {
-        self.cancel_key_deletion_with_context(request, &OperationContext::internal())
-            .await
-    }
-
-    /// Cancel key deletion on behalf of `context`'s principal
-    pub async fn cancel_key_deletion_with_context(
-        &self,
-        request: CancelKeyDeletionRequest,
-        context: &OperationContext,
-    ) -> Result<CancelKeyDeletionResponse> {
-        let started = Instant::now();
-        let key_id = request.key_id.clone();
-        let result = self.cancel_key_deletion_inner(request).await;
-        self.audit(KmsAuditOperation::CancelKeyDeletion, context, Some(&key_id), started, &result);
-        result
-    }
-
-    async fn cancel_key_deletion_inner(&self, request: CancelKeyDeletionRequest) -> Result<CancelKeyDeletionResponse> {
         let response = self.backend.cancel_key_deletion(request).await?;
 
         // Update cache if enabled
@@ -292,60 +159,24 @@ impl KmsManager {
     }
 
     /// Enable a disabled key
-    ///
-    /// Audited as an internal operation; callers serving an authenticated
-    /// request should use [`Self::enable_key_with_context`].
     pub async fn enable_key(&self, key_id: &str) -> Result<()> {
-        self.enable_key_with_context(key_id, &OperationContext::internal()).await
-    }
-
-    /// Enable a disabled key on behalf of `context`'s principal
-    pub async fn enable_key_with_context(&self, key_id: &str, context: &OperationContext) -> Result<()> {
-        let started = Instant::now();
-        let result = self.backend.enable_key(key_id).await;
-        if result.is_ok() {
-            self.invalidate_cached_metadata(key_id).await;
-        }
-        self.audit(KmsAuditOperation::EnableKey, context, Some(key_id), started, &result);
-        result
+        self.backend.enable_key(key_id).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
     }
 
     /// Disable a key; existing data remains decryptable
-    ///
-    /// Audited as an internal operation; callers serving an authenticated
-    /// request should use [`Self::disable_key_with_context`].
     pub async fn disable_key(&self, key_id: &str) -> Result<()> {
-        self.disable_key_with_context(key_id, &OperationContext::internal()).await
-    }
-
-    /// Disable a key on behalf of `context`'s principal
-    pub async fn disable_key_with_context(&self, key_id: &str, context: &OperationContext) -> Result<()> {
-        let started = Instant::now();
-        let result = self.backend.disable_key(key_id).await;
-        if result.is_ok() {
-            self.invalidate_cached_metadata(key_id).await;
-        }
-        self.audit(KmsAuditOperation::DisableKey, context, Some(key_id), started, &result);
-        result
+        self.backend.disable_key(key_id).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
     }
 
     /// Rotate a key to a new version
-    ///
-    /// Audited as an internal operation; callers serving an authenticated
-    /// request should use [`Self::rotate_key_with_context`].
     pub async fn rotate_key(&self, key_id: &str) -> Result<()> {
-        self.rotate_key_with_context(key_id, &OperationContext::internal()).await
-    }
-
-    /// Rotate a key on behalf of `context`'s principal
-    pub async fn rotate_key_with_context(&self, key_id: &str, context: &OperationContext) -> Result<()> {
-        let started = Instant::now();
-        let result = self.backend.rotate_key(key_id).await;
-        if result.is_ok() {
-            self.invalidate_cached_metadata(key_id).await;
-        }
-        self.audit(KmsAuditOperation::RotateKey, context, Some(key_id), started, &result);
-        result
+        self.backend.rotate_key(key_id).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
     }
 
     /// Drop cached metadata after a state mutation so the next describe
@@ -377,470 +208,11 @@ impl KmsManager {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::audit::KmsAuditOutcome;
     use crate::backends::local::LocalKmsBackend;
-    use crate::error::KmsError;
-    use crate::types::{KeyMetadata, KeySpec, KeyState, KeyUsage};
-    use async_trait::async_trait;
-    use base64::Engine as _;
-    use jiff::Zoned;
+    use crate::types::{KeySpec, KeyState, KeyUsage};
     use std::collections::HashMap;
-    use std::sync::Mutex;
     use tempfile::tempdir;
 
-    /// Sink that keeps every record so tests can assert on the audit trail.
-    #[derive(Default)]
-    struct CapturingSink {
-        records: Mutex<Vec<KmsAuditRecord>>,
-    }
-
-    impl KmsAuditSink for CapturingSink {
-        fn emit(&self, record: KmsAuditRecord) {
-            self.records
-                .lock()
-                .expect("audit records lock should not be poisoned")
-                .push(record);
-        }
-    }
-
-    impl CapturingSink {
-        fn records(&self) -> Vec<KmsAuditRecord> {
-            self.records
-                .lock()
-                .expect("audit records lock should not be poisoned")
-                .clone()
-        }
-
-        fn take_one(&self) -> KmsAuditRecord {
-            let mut records = self.records.lock().expect("audit records lock should not be poisoned");
-            assert_eq!(records.len(), 1, "expected exactly one audit record, got {records:?}");
-            records.remove(0)
-        }
-    }
-
-    /// Backend whose management operations all succeed or all fail, so a
-    /// single test can drive every audited operation down both paths —
-    /// including operations no real backend supports on both.
-    struct ScriptedBackend {
-        failure: Option<KmsError>,
-    }
-
-    impl ScriptedBackend {
-        fn succeeding() -> Self {
-            Self { failure: None }
-        }
-
-        fn failing(failure: KmsError) -> Self {
-            Self { failure: Some(failure) }
-        }
-
-        fn check(&self) -> Result<()> {
-            match &self.failure {
-                Some(failure) => Err(failure.clone()),
-                None => Ok(()),
-            }
-        }
-
-        fn metadata(key_id: &str) -> KeyMetadata {
-            KeyMetadata {
-                key_id: key_id.to_string(),
-                key_state: KeyState::Enabled,
-                key_usage: KeyUsage::EncryptDecrypt,
-                description: None,
-                creation_date: Zoned::now(),
-                deletion_date: None,
-                origin: "RUSTFS_KMS".to_string(),
-                key_manager: "RUSTFS".to_string(),
-                tags: HashMap::new(),
-            }
-        }
-    }
-
-    #[async_trait]
-    impl KmsBackend for ScriptedBackend {
-        async fn create_key(&self, request: CreateKeyRequest) -> Result<CreateKeyResponse> {
-            self.check()?;
-            let key_id = request.key_name.unwrap_or_else(|| "scripted-key".to_string());
-            Ok(CreateKeyResponse {
-                key_metadata: Self::metadata(&key_id),
-                key_id,
-            })
-        }
-
-        async fn encrypt(&self, _request: EncryptRequest) -> Result<EncryptResponse> {
-            unimplemented!("data plane is not audited by the manager")
-        }
-
-        async fn decrypt(&self, _request: DecryptRequest) -> Result<DecryptResponse> {
-            unimplemented!("data plane is not audited by the manager")
-        }
-
-        async fn generate_data_key(&self, _request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
-            unimplemented!("data plane is not audited by the manager")
-        }
-
-        async fn describe_key(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
-            self.check()?;
-            Ok(DescribeKeyResponse {
-                key_metadata: Self::metadata(&request.key_id),
-            })
-        }
-
-        async fn list_keys(&self, _request: ListKeysRequest) -> Result<ListKeysResponse> {
-            self.check()?;
-            Ok(ListKeysResponse {
-                keys: Vec::new(),
-                next_marker: None,
-                truncated: false,
-            })
-        }
-
-        async fn delete_key(&self, request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
-            self.check()?;
-            Ok(DeleteKeyResponse {
-                key_metadata: Self::metadata(&request.key_id),
-                key_id: request.key_id,
-                deletion_date: None,
-            })
-        }
-
-        async fn cancel_key_deletion(&self, request: CancelKeyDeletionRequest) -> Result<CancelKeyDeletionResponse> {
-            self.check()?;
-            Ok(CancelKeyDeletionResponse {
-                key_metadata: Self::metadata(&request.key_id),
-                key_id: request.key_id,
-            })
-        }
-
-        async fn enable_key(&self, _key_id: &str) -> Result<()> {
-            self.check()
-        }
-
-        async fn disable_key(&self, _key_id: &str) -> Result<()> {
-            self.check()
-        }
-
-        async fn rotate_key(&self, _key_id: &str) -> Result<()> {
-            self.check()
-        }
-
-        async fn health_check(&self) -> Result<bool> {
-            Ok(true)
-        }
-    }
-
-    const AUDITED_KEY_ID: &str = "audited-key";
-
-    /// Prefix length used when checking that a record embedded no *part* of a
-    /// secret. Long enough that a collision with unrelated text is not a real
-    /// concern.
-    const FRAGMENT_LEN: usize = 24;
-
-    fn scripted_manager(backend: ScriptedBackend) -> (KmsManager, Arc<CapturingSink>) {
-        let temp_dir = tempdir().expect("Failed to create temp dir");
-        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
-        let sink = Arc::new(CapturingSink::default());
-        let manager = KmsManager::new(Arc::new(backend), config).with_audit_sink(sink.clone());
-        (manager, sink)
-    }
-
-    fn request_context() -> OperationContext {
-        OperationContext::new("arn:aws:iam::user/alice".to_string())
-            .with_source_ip("192.0.2.10".to_string())
-            .with_user_agent("rustfs-admin/1".to_string())
-            .with_context("requestID".to_string(), "req-42".to_string())
-    }
-
-    /// Drive one management operation and return the single record it emitted.
-    async fn run_operation(manager: &KmsManager, operation: KmsAuditOperation, context: &OperationContext) -> Result<()> {
-        match operation {
-            KmsAuditOperation::CreateKey => manager
-                .create_key_with_context(
-                    CreateKeyRequest {
-                        key_name: Some(AUDITED_KEY_ID.to_string()),
-                        ..Default::default()
-                    },
-                    context,
-                )
-                .await
-                .map(|_| ()),
-            KmsAuditOperation::DescribeKey => manager
-                .describe_key_with_context(
-                    DescribeKeyRequest {
-                        key_id: AUDITED_KEY_ID.to_string(),
-                    },
-                    context,
-                )
-                .await
-                .map(|_| ()),
-            KmsAuditOperation::ListKeys => manager
-                .list_keys_with_context(ListKeysRequest::default(), context)
-                .await
-                .map(|_| ()),
-            KmsAuditOperation::ScheduleKeyDeletion => manager
-                .delete_key_with_context(
-                    DeleteKeyRequest {
-                        key_id: AUDITED_KEY_ID.to_string(),
-                        pending_window_in_days: None,
-                        force_immediate: None,
-                    },
-                    context,
-                )
-                .await
-                .map(|_| ()),
-            KmsAuditOperation::CancelKeyDeletion => manager
-                .cancel_key_deletion_with_context(
-                    CancelKeyDeletionRequest {
-                        key_id: AUDITED_KEY_ID.to_string(),
-                    },
-                    context,
-                )
-                .await
-                .map(|_| ()),
-            KmsAuditOperation::EnableKey => manager.enable_key_with_context(AUDITED_KEY_ID, context).await,
-            KmsAuditOperation::DisableKey => manager.disable_key_with_context(AUDITED_KEY_ID, context).await,
-            KmsAuditOperation::RotateKey => manager.rotate_key_with_context(AUDITED_KEY_ID, context).await,
-            // Physical removal happens on the background sweep, not here.
-            KmsAuditOperation::DeleteKey => unreachable!("removal is audited by the deletion worker"),
-        }
-    }
-
-    /// Every management operation the manager serves, in audit terms.
-    const AUDITED_OPERATIONS: [KmsAuditOperation; 8] = [
-        KmsAuditOperation::CreateKey,
-        KmsAuditOperation::DescribeKey,
-        KmsAuditOperation::ListKeys,
-        KmsAuditOperation::ScheduleKeyDeletion,
-        KmsAuditOperation::CancelKeyDeletion,
-        KmsAuditOperation::EnableKey,
-        KmsAuditOperation::DisableKey,
-        KmsAuditOperation::RotateKey,
-    ];
-
-    #[tokio::test]
-    async fn every_management_operation_emits_a_complete_success_record() {
-        for operation in AUDITED_OPERATIONS {
-            let (manager, sink) = scripted_manager(ScriptedBackend::succeeding());
-            let context = request_context();
-
-            let outer = Instant::now();
-            run_operation(&manager, operation, &context)
-                .await
-                .unwrap_or_else(|error| panic!("{} should succeed: {error}", operation.as_str()));
-            let outer_elapsed = outer.elapsed();
-
-            let record = sink.take_one();
-            assert_eq!(record.operation, operation);
-            assert_eq!(record.event, operation.event_name());
-            assert_eq!(record.outcome, KmsAuditOutcome::Success);
-            assert_eq!(record.error_class, None);
-            assert_eq!(record.operation_id, context.operation_id);
-            assert_eq!(record.principal, "arn:aws:iam::user/alice");
-            assert_eq!(record.source_ip.as_deref(), Some("192.0.2.10"));
-            assert_eq!(record.user_agent.as_deref(), Some("rustfs-admin/1"));
-            assert_eq!(record.backend, "local");
-            assert_eq!(record.context.get("requestID").map(String::as_str), Some("req-42"));
-            assert!(
-                record.latency <= outer_elapsed,
-                "{} reported a latency larger than the call it measured",
-                operation.as_str()
-            );
-
-            // Listing spans keys; every other operation names the key it touched.
-            if operation == KmsAuditOperation::ListKeys {
-                assert_eq!(record.key_id, None);
-            } else {
-                assert_eq!(record.key_id.as_deref(), Some(AUDITED_KEY_ID));
-            }
-        }
-    }
-
-    #[tokio::test]
-    async fn every_management_operation_emits_a_failure_record() {
-        for operation in AUDITED_OPERATIONS {
-            let (manager, sink) = scripted_manager(ScriptedBackend::failing(KmsError::access_denied("denied by policy")));
-            let context = request_context();
-
-            let error = run_operation(&manager, operation, &context)
-                .await
-                .expect_err("scripted backend should reject the operation");
-            assert!(matches!(error, KmsError::AccessDenied { .. }));
-
-            let record = sink.take_one();
-            assert_eq!(record.operation, operation);
-            assert_eq!(record.event, operation.event_name());
-            assert_eq!(record.outcome, KmsAuditOutcome::Failure);
-            assert_eq!(record.error_class, Some("access_denied"));
-            assert_eq!(record.operation_id, context.operation_id);
-            assert_eq!(record.principal, "arn:aws:iam::user/alice");
-            assert_eq!(record.source_ip.as_deref(), Some("192.0.2.10"));
-
-            // A denied create still has to name the key the caller asked for,
-            // otherwise the record cannot answer "what were they after".
-            if operation != KmsAuditOperation::ListKeys {
-                assert_eq!(record.key_id.as_deref(), Some(AUDITED_KEY_ID));
-            }
-        }
-    }
-
-    #[tokio::test]
-    async fn operations_are_unaffected_when_no_sink_is_installed() {
-        // The audit trail is optional; without a sink the manager must behave
-        // exactly as it did before records existed.
-        let temp_dir = tempdir().expect("Failed to create temp dir");
-        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
-        let manager = KmsManager::new(Arc::new(ScriptedBackend::succeeding()), config);
-
-        for operation in AUDITED_OPERATIONS {
-            run_operation(&manager, operation, &request_context())
-                .await
-                .unwrap_or_else(|error| panic!("{} should succeed without a sink: {error}", operation.as_str()));
-        }
-    }
-
-    #[tokio::test]
-    async fn context_free_calls_are_attributed_to_the_internal_principal() {
-        // Callers that have no authenticated identity must still be
-        // distinguishable from an identity we failed to record.
-        let (manager, sink) = scripted_manager(ScriptedBackend::succeeding());
-
-        manager
-            .describe_key(DescribeKeyRequest {
-                key_id: AUDITED_KEY_ID.to_string(),
-            })
-            .await
-            .expect("describe should succeed");
-
-        let record = sink.take_one();
-        assert_eq!(record.principal, OperationContext::INTERNAL_PRINCIPAL);
-        assert_eq!(record.source_ip, None);
-        assert_eq!(record.user_agent, None);
-    }
-
-    #[tokio::test]
-    async fn unsupported_lifecycle_operations_are_audited_with_their_own_class() {
-        // The local backend has no version history, so rotation is a capability
-        // gap rather than a policy denial; the audit trail must say so.
-        let temp_dir = tempdir().expect("Failed to create temp dir");
-        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
-        let backend = Arc::new(LocalKmsBackend::new(config.clone()).await.expect("Failed to create backend"));
-        let sink = Arc::new(CapturingSink::default());
-        let manager = KmsManager::new(backend, config).with_audit_sink(sink.clone());
-
-        let key_id = manager
-            .create_key_with_context(
-                CreateKeyRequest {
-                    key_name: Some("rotate-me".to_string()),
-                    ..Default::default()
-                },
-                &request_context(),
-            )
-            .await
-            .expect("create should succeed")
-            .key_id;
-
-        manager
-            .rotate_key_with_context(&key_id, &request_context())
-            .await
-            .expect_err("local rotation must be rejected");
-
-        let records = sink.records();
-        let rotate = records.last().expect("rotation should be audited");
-        assert_eq!(rotate.operation, KmsAuditOperation::RotateKey);
-        assert_eq!(rotate.outcome, KmsAuditOutcome::Failure);
-        assert_eq!(rotate.error_class, Some("unsupported_capability"));
-        assert_eq!(rotate.key_id.as_deref(), Some(key_id.as_str()));
-    }
-
-    /// Negative assertion: no audit record may reproduce key material. Driven
-    /// against the real local backend so the assertion covers whatever the
-    /// backend actually hands back, not a hand-written stand-in.
-    #[tokio::test]
-    async fn audit_records_never_reproduce_key_material() {
-        let temp_dir = tempdir().expect("Failed to create temp dir");
-        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
-        let backend = Arc::new(LocalKmsBackend::new(config.clone()).await.expect("Failed to create backend"));
-        let sink = Arc::new(CapturingSink::default());
-        let manager = KmsManager::new(backend, config).with_audit_sink(sink.clone());
-
-        let grant_token = "grant-token-cec4d4b5a1";
-        let context = request_context();
-
-        let key_id = manager
-            .create_key_with_context(
-                CreateKeyRequest {
-                    key_name: Some("material-key".to_string()),
-                    ..Default::default()
-                },
-                &context,
-            )
-            .await
-            .expect("create should succeed")
-            .key_id;
-
-        // Produce real key material, then keep driving the management plane so
-        // any record built afterwards is covered by the assertions below.
-        let data_key = manager
-            .generate_data_key(GenerateDataKeyRequest {
-                key_id: key_id.clone(),
-                key_spec: KeySpec::Aes256,
-                encryption_context: HashMap::from([("bucket".to_string(), "secrets".to_string())]),
-            })
-            .await
-            .expect("data key generation should succeed");
-        let decrypted = manager
-            .decrypt(DecryptRequest {
-                ciphertext: data_key.ciphertext_blob.clone(),
-                encryption_context: HashMap::from([("bucket".to_string(), "secrets".to_string())]),
-                grant_tokens: vec![grant_token.to_string()],
-            })
-            .await
-            .expect("decrypt should succeed");
-
-        manager
-            .describe_key_with_context(DescribeKeyRequest { key_id: key_id.clone() }, &context)
-            .await
-            .expect("describe should succeed");
-        manager
-            .list_keys_with_context(ListKeysRequest::default(), &context)
-            .await
-            .expect("list should succeed");
-        manager
-            .disable_key_with_context(&key_id, &context)
-            .await
-            .expect("disable should succeed");
-        manager
-            .enable_key_with_context(&key_id, &context)
-            .await
-            .expect("enable should succeed");
-
-        let base64 = base64::engine::general_purpose::STANDARD;
-        let encodings = |bytes: &[u8]| vec![hex::encode(bytes), base64.encode(bytes)];
-        let mut forbidden = vec![grant_token.to_string()];
-        forbidden.extend(encodings(&data_key.plaintext_key));
-        forbidden.extend(encodings(&decrypted.plaintext));
-        forbidden.extend(encodings(&data_key.ciphertext_blob));
-        // Fragments catch a record that embedded only part of a blob.
-        let fragments: Vec<String> = forbidden
-            .iter()
-            .filter(|secret| secret.len() > FRAGMENT_LEN)
-            .map(|secret| secret[..FRAGMENT_LEN].to_string())
-            .collect();
-        forbidden.extend(fragments);
-
-        let records = sink.records();
-        assert!(!records.is_empty(), "management operations should have been audited");
-        for record in &records {
-            let rendered = format!("{record:?}");
-            for secret in &forbidden {
-                assert!(
-                    !rendered.contains(secret.as_str()),
-                    "audit record leaked key material or a grant token: {rendered}"
-                );
-            }
-        }
-    }
-
     #[tokio::test]
     async fn test_manager_operations() {
         let temp_dir = tempdir().expect("Failed to create temp dir");

crates/kms/src/service.rs

@@ -112,23 +112,6 @@ impl ObjectEncryptionService {
         self.kms_manager.create_key(request).await
     }
 
-    /// Create a new master key on behalf of `context`'s principal
-    ///
-    /// # Arguments
-    /// * `request` - CreateKeyRequest with key parameters
-    /// * `context` - Identity and correlation data recorded in the audit trail
-    ///
-    /// # Returns
-    /// CreateKeyResponse with created key details
-    ///
-    pub async fn create_key_with_context(
-        &self,
-        request: CreateKeyRequest,
-        context: &OperationContext,
-    ) -> Result<CreateKeyResponse> {
-        self.kms_manager.create_key_with_context(request, context).await
-    }
-
     /// Describe a master key (delegates to KMS manager)
     ///
     /// # Arguments
@@ -141,23 +124,6 @@ impl ObjectEncryptionService {
         self.kms_manager.describe_key(request).await
     }
 
-    /// Describe a master key on behalf of `context`'s principal
-    ///
-    /// # Arguments
-    /// * `request` - DescribeKeyRequest with key ID
-    /// * `context` - Identity and correlation data recorded in the audit trail
-    ///
-    /// # Returns
-    /// DescribeKeyResponse with key metadata
-    ///
-    pub async fn describe_key_with_context(
-        &self,
-        request: DescribeKeyRequest,
-        context: &OperationContext,
-    ) -> Result<DescribeKeyResponse> {
-        self.kms_manager.describe_key_with_context(request, context).await
-    }
-
     /// List master keys (delegates to KMS manager)
     ///
     /// # Arguments
@@ -170,19 +136,6 @@ impl ObjectEncryptionService {
         self.kms_manager.list_keys(request).await
     }
 
-    /// List master keys on behalf of `context`'s principal
-    ///
-    /// # Arguments
-    /// * `request` - ListKeysRequest with listing parameters
-    /// * `context` - Identity and correlation data recorded in the audit trail
-    ///
-    /// # Returns
-    /// ListKeysResponse with list of keys
-    ///
-    pub async fn list_keys_with_context(&self, request: ListKeysRequest, context: &OperationContext) -> Result<ListKeysResponse> {
-        self.kms_manager.list_keys_with_context(request, context).await
-    }
-
     /// Generate a data encryption key (delegates to KMS manager)
     ///
     /// # Arguments

crates/kms/src/service_manager.rs

@@ -14,7 +14,6 @@
 
 //! KMS service manager for dynamic configuration and runtime management
 
-use crate::audit::KmsAuditSink;
 use crate::backends::vault_credentials::CredentialTaskHandle;
 use crate::backends::{KmsBackend, local::LocalKmsBackend};
 use crate::config::{BackendConfig, KmsConfig};
@@ -172,8 +171,6 @@ pub struct KmsServiceManager {
     lifecycle_mutex: Arc<Mutex<()>>,
     /// External reference checker consulted before expired keys are removed
     deletion_reference_checker: std::sync::RwLock<Option<Arc<dyn DeletionReferenceChecker>>>,
-    /// External sink receiving audit records for management operations
-    audit_sink: std::sync::RwLock<Option<Arc<dyn KmsAuditSink>>>,
 }
 
 impl KmsServiceManager {
@@ -188,26 +185,9 @@ impl KmsServiceManager {
             version_counter: Arc::new(AtomicU64::new(0)),
             lifecycle_mutex: Arc::new(Mutex::new(())),
             deletion_reference_checker: std::sync::RwLock::new(None),
-            audit_sink: std::sync::RwLock::new(None),
         }
     }
 
-    /// Install the sink that receives an audit record for every KMS
-    /// management operation. Takes effect for services built by the next
-    /// start or reconfigure.
-    ///
-    /// Without a sink no records are built, so KMS operations are unaffected
-    /// when auditing is not configured.
-    pub fn set_audit_sink(&self, sink: Arc<dyn KmsAuditSink>) {
-        if let Ok(mut slot) = self.audit_sink.write() {
-            *slot = Some(sink);
-        }
-    }
-
-    fn audit_sink(&self) -> Option<Arc<dyn KmsAuditSink>> {
-        self.audit_sink.read().ok().and_then(|slot| slot.clone())
-    }
-
     /// Install the reference checker consulted before the deletion worker
     /// removes an expired key. Takes effect for workers spawned by the next
     /// start or reconfigure.
@@ -605,11 +585,7 @@ impl KmsServiceManager {
         };
 
         // Create KMS manager
-        let mut kms_manager = KmsManager::new(backend, config.clone());
-        if let Some(sink) = self.audit_sink() {
-            kms_manager = kms_manager.with_audit_sink(sink);
-        }
-        let kms_manager = Arc::new(kms_manager);
+        let kms_manager = Arc::new(KmsManager::new(backend, config.clone()));
 
         // Create encryption service
         let encryption_service = Arc::new(ObjectEncryptionService::new((*kms_manager).clone()));
@@ -653,13 +629,7 @@ impl KmsServiceManager {
             return None;
         }
         let cancel = CancellationToken::new();
-        let worker = DeletionWorker::new(
-            backend,
-            config.default_key_id.clone(),
-            self.deletion_reference_checker(),
-            config.backend.as_str(),
-        )
-        .with_audit_sink(self.audit_sink());
+        let worker = DeletionWorker::new(backend, config.default_key_id.clone(), self.deletion_reference_checker());
         let task = worker.spawn(cancel.clone());
         Some(Arc::new(DeletionWorkerHandle {
             cancel,

crates/kms/src/types.rs

@@ -484,20 +484,6 @@ impl OperationContext {
         }
     }
 
-    /// Principal recorded for work the server starts on its own behalf.
-    ///
-    /// Namespaced so it can never collide with an access key or an IAM ARN,
-    /// which keeps "no human did this" distinguishable from "we lost track of
-    /// who did this" in the audit trail.
-    pub const INTERNAL_PRINCIPAL: &'static str = "rustfs:internal";
-
-    /// Context for an operation with no authenticated caller, such as
-    /// background maintenance or a call made while serving a data-plane
-    /// request that carries its own audit entry.
-    pub fn internal() -> Self {
-        Self::new(Self::INTERNAL_PRINCIPAL.to_string())
-    }
-
     /// Add additional context
     ///
     /// # Arguments

crates/notify/src/rules/mod.rs

@@ -22,8 +22,6 @@ mod pattern_rules_test;
 #[cfg(test)]
 mod pattern_test;
 mod rules_map;
-#[cfg(test)]
-mod rules_map_test;
 mod subscriber_index;
 mod subscriber_snapshot;
 mod target_id_set;

crates/notify/src/rules/rules_map_test.rs

@@ -1,104 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Regressions for rule matching against the KMS event namespace.
-//!
-//! KMS events (backlog#1583) are appended to `EventName` for the audit sink,
-//! but they must stay invisible to bucket notification rules: a subscriber
-//! asking for `s3:ObjectCreated:*` — or for everything — must never receive
-//! key management activity.
-
-use super::RulesMap;
-use rustfs_s3_types::EventName;
-use rustfs_targets::arn::TargetID;
-
-const KMS_EVENTS: &[EventName] = &[
-    EventName::KmsKeyCreated,
-    EventName::KmsKeyRotated,
-    EventName::KmsKeyEnabled,
-    EventName::KmsKeyDisabled,
-    EventName::KmsKeyDeletionScheduled,
-    EventName::KmsKeyDeletionCancelled,
-    EventName::KmsKeyDeleted,
-    EventName::KmsKeyAccessed,
-];
-
-fn test_target() -> TargetID {
-    TargetID::new("primary".to_string(), "webhook".to_string())
-}
-
-fn rules_for(events: &[EventName]) -> RulesMap {
-    let mut rules = RulesMap::new();
-    rules.add_rule_config(events, String::new(), test_target());
-    rules
-}
-
-#[test]
-fn s3_wildcard_subscriptions_do_not_match_kms_events() {
-    let rules = rules_for(&[
-        EventName::ObjectCreatedAll,
-        EventName::ObjectAccessedAll,
-        EventName::ObjectRemovedAll,
-        EventName::ObjectTaggingAll,
-        EventName::ObjectReplicationAll,
-        EventName::ObjectRestoreAll,
-        EventName::ObjectTransitionAll,
-        EventName::LifecycleExpirationAll,
-        EventName::ObjectScannerAll,
-    ]);
-
-    for event in KMS_EVENTS {
-        assert!(!rules.has_subscriber(event), "{event} must not have an S3 wildcard subscriber");
-        assert!(
-            rules.match_rules(*event, "any/object").is_empty(),
-            "{event} must not match any S3 wildcard rule"
-        );
-    }
-}
-
-#[test]
-fn everything_subscription_does_not_match_kms_events() {
-    let rules = rules_for(&[EventName::Everything]);
-
-    // Sanity: the catch-all really is subscribed to the S3 surface.
-    assert!(rules.has_subscriber(&EventName::ObjectCreatedPut));
-
-    for event in KMS_EVENTS {
-        assert!(!rules.has_subscriber(event), "{event} must stay outside the s3 catch-all");
-        assert!(
-            rules.match_rules(*event, "any/object").is_empty(),
-            "{event} must not match the s3 catch-all rule"
-        );
-    }
-}
-
-#[test]
-fn kms_subscription_does_not_leak_into_s3_matching() {
-    // The inverse direction: even an explicit KMS subscription must not widen
-    // the mask so that S3 events start matching a KMS-only rule.
-    let rules = rules_for(KMS_EVENTS);
-
-    for event in [
-        EventName::ObjectCreatedPut,
-        EventName::ObjectRemovedDelete,
-        EventName::ObjectAccessedGet,
-        EventName::BucketCreated,
-    ] {
-        assert!(!rules.has_subscriber(&event), "{event} must not match a KMS-only rule");
-        assert!(
-            rules.match_rules(event, "any/object").is_empty(),
-            "{event} must not match a KMS-only rule"
-        );
-    }
-}

crates/s3-types/src/event_name.rs

@@ -86,20 +86,6 @@ pub enum EventName {
     ObjectRemovedAbortMultipartUpload,
     ObjectCreatedCreateMultipartUpload,
     ObjectRemovedDeleteObjects,
-
-    // KMS management-plane events. They travel to the audit sink only and are
-    // never produced by the bucket notification path, so no compound `s3:`
-    // event expands to them. New variants must keep being appended here: the
-    // discriminant of every preceding variant is a `mask()` bit position, and
-    // inserting in the middle would silently renumber existing bits.
-    KmsKeyCreated,
-    KmsKeyRotated,
-    KmsKeyEnabled,
-    KmsKeyDisabled,
-    KmsKeyDeletionScheduled,
-    KmsKeyDeletionCancelled,
-    KmsKeyDeleted,
-    KmsKeyAccessed,
 }
 
 // Single event type sequential array for Everything.expand()
@@ -200,17 +186,6 @@ impl EventName {
             "s3:Scanner:LargeVersions" => Ok(EventName::ScannerLargeVersions),
             "s3:Scanner:BigPrefix" => Ok(EventName::ScannerBigPrefix),
             "s3:Scanner:*" => Ok(EventName::ObjectScannerAll),
-            // KMS events use their own namespace so a `s3:` wildcard in a bucket
-            // notification config can never select them. They still round-trip
-            // because audit entries are persisted and replayed by the store targets.
-            "kms:Key:Created" => Ok(EventName::KmsKeyCreated),
-            "kms:Key:Rotated" => Ok(EventName::KmsKeyRotated),
-            "kms:Key:Enabled" => Ok(EventName::KmsKeyEnabled),
-            "kms:Key:Disabled" => Ok(EventName::KmsKeyDisabled),
-            "kms:Key:DeletionScheduled" => Ok(EventName::KmsKeyDeletionScheduled),
-            "kms:Key:DeletionCancelled" => Ok(EventName::KmsKeyDeletionCancelled),
-            "kms:Key:Deleted" => Ok(EventName::KmsKeyDeleted),
-            "kms:Key:Accessed" => Ok(EventName::KmsKeyAccessed),
             // `Everything` has no string representation (`as_str` yields ""), so it
             // cannot be parsed back from a string. Every other variant round-trips.
             _ => Err(ParseEventNameError(s.to_string())),
@@ -276,14 +251,6 @@ impl EventName {
             EventName::ObjectRemovedAbortMultipartUpload => "s3:ObjectRemoved:AbortMultipartUpload",
             EventName::ObjectCreatedCreateMultipartUpload => "s3:ObjectCreated:CreateMultipartUpload",
             EventName::ObjectRemovedDeleteObjects => "s3:ObjectRemoved:DeleteObjects",
-            EventName::KmsKeyCreated => "kms:Key:Created",
-            EventName::KmsKeyRotated => "kms:Key:Rotated",
-            EventName::KmsKeyEnabled => "kms:Key:Enabled",
-            EventName::KmsKeyDisabled => "kms:Key:Disabled",
-            EventName::KmsKeyDeletionScheduled => "kms:Key:DeletionScheduled",
-            EventName::KmsKeyDeletionCancelled => "kms:Key:DeletionCancelled",
-            EventName::KmsKeyDeleted => "kms:Key:Deleted",
-            EventName::KmsKeyAccessed => "kms:Key:Accessed",
         }
     }
 
@@ -390,25 +357,6 @@ impl EventName {
                 | EventName::ObjectRemovedDeleteObjects
         )
     }
-
-    /// Returns `true` for KMS management-plane events.
-    ///
-    /// These are audit-only: they are never emitted through the bucket
-    /// notification pipeline, and no `s3:` event selector expands to them.
-    #[inline]
-    pub fn is_kms(&self) -> bool {
-        matches!(
-            self,
-            EventName::KmsKeyCreated
-                | EventName::KmsKeyRotated
-                | EventName::KmsKeyEnabled
-                | EventName::KmsKeyDisabled
-                | EventName::KmsKeyDeletionScheduled
-                | EventName::KmsKeyDeletionCancelled
-                | EventName::KmsKeyDeleted
-                | EventName::KmsKeyAccessed
-        )
-    }
 }
 
 /// Returns the S3 notification event schema version for a given event.
@@ -622,26 +570,6 @@ mod tests {
         EventName::ObjectRemovedAbortMultipartUpload,
         EventName::ObjectCreatedCreateMultipartUpload,
         EventName::ObjectRemovedDeleteObjects,
-        EventName::KmsKeyCreated,
-        EventName::KmsKeyRotated,
-        EventName::KmsKeyEnabled,
-        EventName::KmsKeyDisabled,
-        EventName::KmsKeyDeletionScheduled,
-        EventName::KmsKeyDeletionCancelled,
-        EventName::KmsKeyDeleted,
-        EventName::KmsKeyAccessed,
-    ];
-
-    /// Every KMS management-plane event.
-    const KMS_EVENT_NAMES: &[EventName] = &[
-        EventName::KmsKeyCreated,
-        EventName::KmsKeyRotated,
-        EventName::KmsKeyEnabled,
-        EventName::KmsKeyDisabled,
-        EventName::KmsKeyDeletionScheduled,
-        EventName::KmsKeyDeletionCancelled,
-        EventName::KmsKeyDeleted,
-        EventName::KmsKeyAccessed,
     ];
 
     /// Regression for backlog#965: `mask()` used to recurse forever for the
@@ -754,54 +682,6 @@ mod tests {
         }
     }
 
-    /// KMS events are audit-plane only. No `s3:` selector — including the
-    /// catch-all `Everything` and every compound "All" type — may share a bit
-    /// with them, otherwise a bucket notification rule would silently start
-    /// matching KMS activity.
-    #[test]
-    fn test_kms_event_masks_are_disjoint_from_every_s3_selector() {
-        let s3_selectors: Vec<EventName> = ALL_EVENT_NAMES.iter().copied().filter(|ev| !ev.is_kms()).collect();
-
-        let mut seen = 0u64;
-        for kms in KMS_EVENT_NAMES {
-            let mask = kms.mask();
-            assert_ne!(mask, 0, "KMS event {kms} must have a non-zero mask");
-            assert_eq!(seen & mask, 0, "KMS event {kms} mask overlaps another KMS event");
-            seen |= mask;
-
-            for s3 in &s3_selectors {
-                assert_eq!(s3.mask() & mask, 0, "KMS event {kms} mask collides with S3 selector {s3}");
-            }
-        }
-    }
-
-    /// KMS event names must live in their own namespace so that neither a
-    /// `s3:` prefix filter nor an `s3:...:*` wildcard can select them.
-    #[test]
-    fn test_kms_event_names_are_outside_the_s3_namespace() {
-        for ev in KMS_EVENT_NAMES {
-            assert!(ev.is_kms(), "{ev} should be classified as a KMS event");
-            assert!(ev.as_str().starts_with("kms:"), "unexpected KMS event name {:?}", ev.as_str());
-            assert_eq!(EventName::parse(ev.as_str()).as_ref(), Ok(ev), "KMS event {ev} must round-trip");
-            assert_eq!(ev.expand(), vec![*ev], "KMS event {ev} must expand to itself only");
-        }
-
-        for ev in ALL_EVENT_NAMES.iter().filter(|ev| !ev.is_kms()) {
-            assert!(!ev.as_str().starts_with("kms:"), "{ev} must not claim the KMS namespace");
-        }
-    }
-
-    /// `mask()` shifts by `discriminant - 1`, so the enum may hold at most 64
-    /// mask-bearing variants. Appending past that overflows the shift instead
-    /// of failing loudly, so keep the budget check next to the variants.
-    #[test]
-    fn test_mask_bit_budget_is_not_exhausted() {
-        for ev in ALL_EVENT_NAMES {
-            let value = *ev as u32;
-            assert!(value <= 64, "{ev} has discriminant {value}; mask() only has 64 bits to hand out");
-        }
-    }
-
     /// `Everything` must cover every sequential single-type bit.
     #[test]
     fn test_everything_mask_covers_all_single_types() {

rustfs/src/main.rs

@@ -12,9 +12,32 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
+use std::alloc::{GlobalAlloc, Layout};
+
+#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
+#[derive(Default)]
+struct DefaultMiMalloc;
+
+#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
+// SAFETY: allocation and deallocation are forwarded unchanged to MiMalloc, so
+// MiMalloc's GlobalAlloc guarantees apply to every returned pointer and layout.
+#[allow(unsafe_code)]
+unsafe impl GlobalAlloc for DefaultMiMalloc {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        // SAFETY: the caller upholds GlobalAlloc's contract for layout.
+        unsafe { mimalloc::MiMalloc.alloc(layout) }
+    }
+
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        // SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
+        unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
+    }
+}
+
 #[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
 #[global_allocator]
-static GLOBAL: hotpath::CountingAllocator = hotpath::CountingAllocator::new();
+static GLOBAL: hotpath::CountingAllocator<DefaultMiMalloc> = hotpath::CountingAllocator::new();
 
 #[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
 #[global_allocator]
@@ -25,3 +48,15 @@ fn main() {
 
     rustfs::startup_entrypoint::run_process();
 }
+
+#[cfg(all(test, feature = "hotpath", feature = "hotpath-alloc"))]
+mod tests {
+    #[test]
+    #[allow(unsafe_code)]
+    fn hotpath_allocator_uses_mimalloc() {
+        let allocation = Box::new([0_u8; 64]);
+
+        // SAFETY: the live Box pointer is valid to inspect for heap ownership.
+        assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
+    }
+}

scripts/check_logging_guardrails.sh

@@ -27,9 +27,6 @@ checked_files=(
   "rustfs/src/storage/rpc/http_service.rs"
   "rustfs/src/storage/rpc/node_service.rs"
   "crates/kms/src/config.rs"
-  "crates/kms/src/audit.rs"
-  "crates/kms/src/manager.rs"
-  "crates/kms/src/deletion_worker.rs"
   "crates/audit/src/pipeline.rs"
   "crates/audit/src/system.rs"
   "crates/audit/src/global.rs"