ci: stop running expensive jobs that cannot inform the result (#5528)

Three independent fixes that all avoid burning self-hosted runners on work
whose outcome is already determined. None of them changes what is tested.

- ci-docs-only: add a "Quick Checks" companion job. It is a prerequisite for
  gating ci.yml's expensive jobs behind quick-checks (rustfs/backlog#1599):
  once "Quick Checks" is a required check, a docs-only PR would otherwise wait
  on it forever. The steps are a byte-identical copy of ci.yml's quick-checks
  rather than an `echo`, so that on a mixed PR the two same-named check runs
  execute the same commands against the same merge ref and cannot disagree —
  GitHub has no written contract for how it picks between same-named required
  check runs, and the real job only takes 45-51s, leaving no timing margin to
  rely on.

- ci: guard uring-integration with the same `closed` check every other job
  already has. The pull_request trigger includes `closed` only so the
  concurrency group cancels in-flight runs; this job had no guard and no
  `needs`, so every closed or merged PR ran the full io_uring suite (4m17s,
  7m19s and 7m31s on runs 30678272341, 30678117601 and 30662728539).

- ci: gate s3-lifecycle-behavior-tests on e2e-tests, matching
  s3-implemented-tests. Both lanes only download the prebuilt debug binary, and
  s3-implemented-tests already finishes later, so a green PR's wall clock is
  unchanged; a red one stops holding a sm-standard-4 for up to 30 minutes.

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

@@ -448,6 +448,13 @@ 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'
     # 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 +753,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"