fix(kms): remove stale local export test import

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

Cargo.lock

@@ -3672,6 +3672,7 @@ dependencies = [
  "flate2",
  "futures",
  "hex",
+ "hotpath",
  "http 1.5.0",
  "http-body-util",
  "hyper",
@@ -9039,6 +9040,7 @@ dependencies = [
  "const-str",
  "futures",
  "hashbrown 0.17.1",
+ "hotpath",
  "metrics",
  "rustfs-config",
  "rustfs-s3-types",
@@ -9059,6 +9061,7 @@ dependencies = [
  "base64-simd",
  "bytes",
  "crc-fast",
+ "hotpath",
  "http 1.5.0",
  "md-5 0.11.0",
  "pretty_assertions",
@@ -9072,6 +9075,7 @@ name = "rustfs-common"
 version = "1.0.0-beta.12"
 dependencies = [
  "chrono",
+ "hotpath",
  "metrics",
  "rmp-serde",
  "s3s",
@@ -9086,6 +9090,7 @@ dependencies = [
 name = "rustfs-concurrency"
 version = "1.0.0-beta.12"
 dependencies = [
+ "hotpath",
  "insta",
  "rustfs-io-core",
  "serde",
@@ -9099,6 +9104,7 @@ name = "rustfs-config"
 version = "1.0.0-beta.12"
 dependencies = [
  "const-str",
+ "hotpath",
  "serde",
  "serde_json",
 ]
@@ -9109,6 +9115,7 @@ version = "1.0.0-beta.12"
 dependencies = [
  "base64-simd",
  "hmac 0.13.0",
+ "hotpath",
  "rand 0.10.2",
  "serde",
  "serde_json",
@@ -9124,6 +9131,7 @@ dependencies = [
  "argon2",
  "base64-simd",
  "chacha20poly1305",
+ "hotpath",
  "jsonwebtoken 11.0.0",
  "pbkdf2 0.13.0",
  "rand 0.10.2",
@@ -9141,6 +9149,7 @@ name = "rustfs-data-usage"
 version = "1.0.0-beta.12"
 dependencies = [
  "async-trait",
+ "hotpath",
  "rmp-serde",
  "rustfs-filemeta",
  "serde",
@@ -9286,6 +9295,7 @@ dependencies = [
 name = "rustfs-extension-schema"
 version = "1.0.0-beta.12"
 dependencies = [
+ "hotpath",
  "serde",
  "serde_json",
  "thiserror 2.0.19",
@@ -9324,6 +9334,7 @@ dependencies = [
  "async-trait",
  "base64 0.23.0",
  "futures",
+ "hotpath",
  "http 1.5.0",
  "metrics",
  "rustfs-common",
@@ -9355,6 +9366,7 @@ dependencies = [
  "async-trait",
  "base64-simd",
  "futures",
+ "hotpath",
  "http 1.5.0",
  "jsonwebtoken 11.0.0",
  "moka",
@@ -9389,6 +9401,7 @@ name = "rustfs-io-core"
 version = "1.0.0-beta.12"
 dependencies = [
  "bytes",
+ "hotpath",
  "memmap2",
  "rustfs-io-metrics",
  "thiserror 2.0.19",
@@ -9401,6 +9414,7 @@ name = "rustfs-io-metrics"
 version = "1.0.0-beta.12"
 dependencies = [
  "criterion",
+ "hotpath",
  "metrics",
  "metrics-util",
  "num_cpus",
@@ -9467,6 +9481,7 @@ version = "1.0.0-beta.12"
 dependencies = [
  "bytes",
  "futures",
+ "hotpath",
  "http 1.5.0",
  "http-body 1.1.0",
  "http-body-util",
@@ -9499,6 +9514,7 @@ dependencies = [
  "base64 0.23.0",
  "chacha20poly1305",
  "hex",
+ "hotpath",
  "insta",
  "jiff",
  "md-5 0.11.0",
@@ -9531,6 +9547,7 @@ name = "rustfs-lifecycle"
 version = "1.0.0-beta.12"
 dependencies = [
  "async-trait",
+ "hotpath",
  "metrics",
  "metrics-util",
  "proptest",
@@ -9555,6 +9572,7 @@ dependencies = [
  "async-trait",
  "crossbeam-queue",
  "futures",
+ "hotpath",
  "parking_lot",
  "rand 0.10.2",
  "rustfs-io-metrics",
@@ -9576,6 +9594,7 @@ version = "1.0.0-beta.12"
 dependencies = [
  "chrono",
  "flate2",
+ "hotpath",
  "regex",
  "serde",
  "serde_json",
@@ -9593,6 +9612,7 @@ name = "rustfs-madmin"
 version = "1.0.0-beta.12"
 dependencies = [
  "chrono",
+ "hotpath",
  "humantime",
  "hyper",
  "rmp-serde",
@@ -9613,6 +9633,7 @@ dependencies = [
  "criterion",
  "form_urlencoded",
  "hashbrown 0.17.1",
+ "hotpath",
  "metrics",
  "percent-encoding",
  "quick-xml",
@@ -9642,6 +9663,7 @@ version = "1.0.0-beta.12"
 dependencies = [
  "criterion",
  "futures",
+ "hotpath",
  "rustfs-config",
  "rustfs-io-metrics",
  "rustfs-utils",
@@ -9660,6 +9682,7 @@ version = "1.0.0-beta.12"
 dependencies = [
  "bytes",
  "criterion",
+ "hotpath",
  "metrics",
  "metrics-util",
  "moka",
@@ -9682,6 +9705,7 @@ dependencies = [
  "flate2",
  "futures-util",
  "glob",
+ "hotpath",
  "jiff",
  "libc",
  "metrics",
@@ -9767,6 +9791,7 @@ dependencies = [
  "futures-util",
  "hex",
  "hmac 0.13.0",
+ "hotpath",
  "http 1.5.0",
  "http-body-util",
  "hyper",
@@ -9819,6 +9844,7 @@ name = "rustfs-protos"
 version = "1.0.0-beta.12"
 dependencies = [
  "flatbuffers",
+ "hotpath",
  "prost 0.14.4",
  "rmp-serde",
  "rustfs-common",
@@ -9843,6 +9869,7 @@ version = "1.0.0-beta.12"
 dependencies = [
  "byteorder",
  "bytes",
+ "hotpath",
  "regex",
  "rmp",
  "rmp-serde",
@@ -9901,6 +9928,7 @@ dependencies = [
  "chacha20poly1305",
  "hex",
  "hmac 0.13.0",
+ "hotpath",
  "minlz",
  "pin-project-lite",
  "rand 0.10.2",
@@ -9918,6 +9946,7 @@ dependencies = [
 name = "rustfs-s3-ops"
 version = "1.0.0-beta.12"
 dependencies = [
+ "hotpath",
  "rustfs-s3-types",
 ]
 
@@ -9925,6 +9954,7 @@ dependencies = [
 name = "rustfs-s3-types"
 version = "1.0.0-beta.12"
 dependencies = [
+ "hotpath",
  "serde",
  "serde_json",
 ]
@@ -9939,6 +9969,7 @@ dependencies = [
  "datafusion",
  "futures",
  "futures-core",
+ "hotpath",
  "http 1.5.0",
  "metrics",
  "parking_lot",
@@ -9967,6 +9998,7 @@ dependencies = [
  "datafusion",
  "derive_builder",
  "futures",
+ "hotpath",
  "parking_lot",
  "rustfs-s3select-api",
  "s3s",
@@ -9984,6 +10016,7 @@ dependencies = [
  "futures",
  "hex-simd",
  "hmac 0.13.0",
+ "hotpath",
  "http 1.5.0",
  "metrics",
  "rand 0.10.2",
@@ -9994,6 +10027,7 @@ dependencies = [
  "rustfs-data-usage",
  "rustfs-ecstore",
  "rustfs-filemeta",
+ "rustfs-lock",
  "rustfs-storage-api",
  "rustfs-utils",
  "s3s",
@@ -10016,6 +10050,7 @@ dependencies = [
 name = "rustfs-security-governance"
 version = "1.0.0-beta.12"
 dependencies = [
+ "hotpath",
  "thiserror 2.0.19",
 ]
 
@@ -10025,6 +10060,7 @@ version = "1.0.0-beta.12"
 dependencies = [
  "base64-simd",
  "bytes",
+ "hotpath",
  "http 1.5.0",
  "hyper",
  "rustfs-utils",
@@ -10041,6 +10077,7 @@ name = "rustfs-storage-api"
 version = "1.0.0-beta.12"
 dependencies = [
  "async-trait",
+ "hotpath",
  "insta",
  "rustfs-filemeta",
  "serde",
@@ -10062,6 +10099,7 @@ dependencies = [
  "deadpool-postgres",
  "futures-util",
  "hashbrown 0.17.1",
+ "hotpath",
  "hyper",
  "hyper-rustls",
  "lapin",
@@ -10107,6 +10145,7 @@ dependencies = [
 name = "rustfs-test-utils"
 version = "1.0.0-beta.12"
 dependencies = [
+ "hotpath",
  "rustfs-data-usage",
  "rustfs-ecstore",
  "rustfs-storage-api",
@@ -10123,6 +10162,7 @@ name = "rustfs-tls-runtime"
 version = "1.0.0-beta.12"
 dependencies = [
  "arc-swap",
+ "hotpath",
  "metrics",
  "rcgen",
  "rustfs-common",
@@ -10144,6 +10184,7 @@ version = "1.0.0-beta.12"
 dependencies = [
  "async-trait",
  "axum",
+ "hotpath",
  "http 1.5.0",
  "ipnetwork",
  "metrics",
@@ -10190,6 +10231,7 @@ dependencies = [
  "hex-simd",
  "highway",
  "hmac 0.13.0",
+ "hotpath",
  "http 1.5.0",
  "hyper",
  "local-ip-address",
@@ -10222,6 +10264,7 @@ dependencies = [
  "astral-tokio-tar",
  "async-compression",
  "criterion",
+ "hotpath",
  "tempfile",
  "thiserror 2.0.19",
  "tokio",

crates/audit/Cargo.toml

@@ -25,7 +25,33 @@ documentation = "https://docs.rs/rustfs-audit/latest/rustfs_audit/"
 keywords = ["audit", "target", "management", "fan-out", "RustFS"]
 categories = ["web-programming", "development-tools", "asynchronous", "api-bindings"]
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "rustfs-config/hotpath",
+    "rustfs-s3-types/hotpath",
+    "rustfs-targets/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-s3-types/hotpath-alloc",
+    "rustfs-targets/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-s3-types/hotpath-cpu",
+    "rustfs-targets/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-targets = { workspace = true }
 rustfs-config = { workspace = true, features = ["audit", "server-config-model"] }
 rustfs-s3-types = { workspace = true }

crates/checksums/Cargo.toml

@@ -28,7 +28,14 @@ documentation = "https://docs.rs/rustfs-checksums/latest/rustfs_checksum/"
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 bytes = { workspace = true, features = ["serde"] }
 crc-fast = { workspace = true }
 http = { workspace = true }

crates/common/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["web-programming", "development-tools", "data-structures"]
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "hotpath/tokio"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 tokio = { workspace = true, features = ["fs", "rt-multi-thread"] }
 tonic = { workspace = true, features = ["gzip", "deflate"] }
 uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] }

crates/concurrency/Cargo.toml

@@ -13,7 +13,14 @@ categories = ["concurrency", "filesystem"]
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-io-core/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-io-core/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-io-core/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 # Internal crates
 rustfs-io-core = { workspace = true }
 serde = { workspace = true, features = ["derive"] }

crates/config/Cargo.toml

@@ -25,6 +25,7 @@ keywords = ["configuration", "settings", "management", "rustfs", "Minio"]
 categories = ["web-programming", "development-tools", "config"]
 
 [dependencies]
+hotpath.workspace = true
 const-str = { workspace = true, optional = true, features = ["std", "proc"] }
 serde = { workspace = true, optional = true, features = ["derive"] }
 serde_json = { workspace = true, optional = true, features = ["raw_value"] }
@@ -34,6 +35,9 @@ workspace = true
 
 [features]
 default = ["constants"]
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 audit = ["dep:const-str", "constants"]
 constants = ["dep:const-str"]
 notify = ["dep:const-str", "constants"]

crates/credentials/Cargo.toml

@@ -24,7 +24,14 @@ description = "Credentials management utilities for RustFS, enabling secure hand
 keywords = ["rustfs", "Minio", "credentials", "authentication", "authorization"]
 categories = ["web-programming", "development-tools", "data-structures", "security"]
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 base64-simd = { workspace = true }
 hmac = { workspace = true }
 rand = { workspace = true, features = ["serde"] }

crates/crypto/Cargo.toml

@@ -29,6 +29,7 @@ documentation = "https://docs.rs/rustfs-crypto/latest/rustfs_crypto/"
 workspace = true
 
 [dependencies]
+hotpath.workspace = true
 aes-gcm = { workspace = true, optional = true, features = ["rand_core"] }
 argon2 = { workspace = true, optional = true }
 chacha20poly1305 = { workspace = true, optional = true }
@@ -49,6 +50,9 @@ time = { workspace = true, features = ["parsing", "formatting", "macros", "serde
 
 [features]
 default = ["crypto", "fips"]
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 fips = []
 crypto = [
     "dep:aes-gcm",

crates/data-usage/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["data-structures", "filesystem"]
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "rustfs-filemeta/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-filemeta/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-filemeta/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 serde = { workspace = true, features = ["derive"] }
 rmp-serde = { workspace = true }
 async-trait = { workspace = true }

crates/e2e_test/Cargo.toml

@@ -25,10 +25,58 @@ workspace = true
 
 [features]
 default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/reqwest-0-13",
+    "rustfs-config/hotpath",
+    "rustfs-credentials/hotpath",
+    "rustfs-data-usage/hotpath",
+    "rustfs-ecstore/hotpath",
+    "rustfs-filemeta/hotpath",
+    "rustfs-lock/hotpath",
+    "rustfs-madmin/hotpath",
+    "rustfs-protos/hotpath",
+    "rustfs-rio/hotpath",
+    "rustfs-signer/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-credentials/hotpath-alloc",
+    "rustfs-data-usage/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-filemeta/hotpath-alloc",
+    "rustfs-lock/hotpath-alloc",
+    "rustfs-madmin/hotpath-alloc",
+    "rustfs-protos/hotpath-alloc",
+    "rustfs-rio/hotpath-alloc",
+    "rustfs-signer/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-credentials/hotpath-cpu",
+    "rustfs-data-usage/hotpath-cpu",
+    "rustfs-ecstore/hotpath-cpu",
+    "rustfs-filemeta/hotpath-cpu",
+    "rustfs-lock/hotpath-cpu",
+    "rustfs-madmin/hotpath-cpu",
+    "rustfs-protos/hotpath-cpu",
+    "rustfs-rio/hotpath-cpu",
+    "rustfs-signer/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
 ftps = []
 sftp = []
 
 [dependencies]
+hotpath.workspace = true
 rustfs-config = { workspace = true, features = ["constants"] }
 rustfs-credentials.workspace = true
 rustfs-ecstore.workspace = true

crates/e2e_test/src/delete_marker_migration_semantics_test.rs

@@ -56,21 +56,6 @@ mod tests {
         );
     }
 
-    async fn assert_current_list_hides_delete_marker(client: &Client, bucket: &str, key: &str) {
-        let listed = client
-            .list_objects_v2()
-            .bucket(bucket)
-            .prefix(key)
-            .send()
-            .await
-            .expect("list current objects after delete marker");
-
-        assert!(
-            listed.contents().iter().all(|object| object.key() != Some(key)),
-            "ListObjectsV2 must hide an object whose latest version is a delete marker"
-        );
-    }
-
     #[tokio::test]
     #[serial]
     async fn test_versioning_only_delete_marker_has_minio_compatible_visibility_for_migration_proof() {
@@ -109,7 +94,6 @@ mod tests {
         assert_eq!(markers[0].version_id(), Some(delete_marker_version_id));
         assert_eq!(markers[0].is_latest(), Some(true));
         assert_current_get_is_delete_marker_not_found(&client, bucket, key).await;
-        assert_current_list_hides_delete_marker(&client, bucket, key).await;
     }
 
     #[tokio::test]
@@ -134,17 +118,6 @@ mod tests {
             .await
             .expect("put historical version");
         let data_version_id = put.version_id().expect("put should return data version id");
-        let listed_before_delete = client
-            .list_objects_v2()
-            .bucket(bucket)
-            .prefix(key)
-            .send()
-            .await
-            .expect("list current object before creating delete marker");
-        assert!(
-            listed_before_delete.contents().iter().any(|object| object.key() == Some(key)),
-            "ListObjectsV2 must include the current object before it is deleted"
-        );
 
         let delete_marker = client
             .delete_object()
@@ -172,7 +145,6 @@ mod tests {
         assert_eq!(markers[0].version_id(), Some(delete_marker_version_id));
         assert_eq!(markers[0].is_latest(), Some(true));
         assert_current_get_is_delete_marker_not_found(&client, bucket, key).await;
-        assert_current_list_hides_delete_marker(&client, bucket, key).await;
 
         let historical = client
             .get_object()

crates/ecstore/Cargo.toml

@@ -40,19 +40,83 @@ hotpath = [
     "hotpath/async-channel",
     "hotpath/parking_lot",
     "hotpath/reqwest-0-13",
+    "rustfs-checksums/hotpath",
+    "rustfs-common/hotpath",
+    "rustfs-concurrency/hotpath",
+    "rustfs-config/hotpath",
+    "rustfs-credentials/hotpath",
+    "rustfs-data-usage/hotpath",
     "rustfs-filemeta/hotpath",
+    "rustfs-io-metrics/hotpath",
+    "rustfs-lifecycle/hotpath",
+    "rustfs-lock/hotpath",
+    "rustfs-madmin/hotpath",
+    "rustfs-object-capacity/hotpath",
+    "rustfs-policy/hotpath",
+    "rustfs-protos/hotpath",
+    "rustfs-replication/hotpath",
     "rustfs-rio/hotpath",
+    "rustfs-rio-v2?/hotpath",
+    "rustfs-s3-types/hotpath",
+    "rustfs-signer/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-tls-runtime/hotpath",
+    "rustfs-utils/hotpath",
+    "rustfs-crypto/hotpath",
 ]
 hotpath-alloc = [
+    "hotpath",
     "hotpath/hotpath-alloc",
+    "rustfs-checksums/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-concurrency/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-credentials/hotpath-alloc",
+    "rustfs-data-usage/hotpath-alloc",
     "rustfs-filemeta/hotpath-alloc",
+    "rustfs-io-metrics/hotpath-alloc",
+    "rustfs-lifecycle/hotpath-alloc",
+    "rustfs-lock/hotpath-alloc",
+    "rustfs-madmin/hotpath-alloc",
+    "rustfs-object-capacity/hotpath-alloc",
+    "rustfs-policy/hotpath-alloc",
+    "rustfs-protos/hotpath-alloc",
+    "rustfs-replication/hotpath-alloc",
     "rustfs-rio/hotpath-alloc",
+    "rustfs-rio-v2?/hotpath-alloc",
+    "rustfs-s3-types/hotpath-alloc",
+    "rustfs-signer/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-tls-runtime/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+    "rustfs-crypto/hotpath-alloc",
 ]
 hotpath-cpu = [
     "hotpath",
     "hotpath/hotpath-cpu",
+    "rustfs-checksums/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-concurrency/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-credentials/hotpath-cpu",
+    "rustfs-data-usage/hotpath-cpu",
     "rustfs-filemeta/hotpath-cpu",
+    "rustfs-io-metrics/hotpath-cpu",
+    "rustfs-lifecycle/hotpath-cpu",
+    "rustfs-lock/hotpath-cpu",
+    "rustfs-madmin/hotpath-cpu",
+    "rustfs-object-capacity/hotpath-cpu",
+    "rustfs-policy/hotpath-cpu",
+    "rustfs-protos/hotpath-cpu",
+    "rustfs-replication/hotpath-cpu",
     "rustfs-rio/hotpath-cpu",
+    "rustfs-rio-v2?/hotpath-cpu",
+    "rustfs-s3-types/hotpath-cpu",
+    "rustfs-signer/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-tls-runtime/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+    "rustfs-crypto/hotpath-cpu",
 ]
 # Exposes shared lifecycle/tier test utilities (MockWarmBackend, fault
 # injection, xl.meta transition assertions) via `api::tier::test_util`.

crates/ecstore/src/erasure/coding/decode.rs

@@ -1645,15 +1645,28 @@ impl Erasure {
             }
             Err(e) => {
                 record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_EMIT, emit_stage_start);
-                error!(
-                    block_offset,
-                    block_length,
-                    bytes_written = *written,
-                    stage = GET_STAGE_EMIT,
-                    reason = classify_io_error(&e).as_str(),
-                    error = ?e,
-                    "Erasure decode failed to emit reconstructed data"
-                );
+                let reason = classify_io_error(&e);
+                if reason == GetObjectFailureReason::DownstreamClosed {
+                    debug!(
+                        block_offset,
+                        block_length,
+                        bytes_written = *written,
+                        stage = GET_STAGE_EMIT,
+                        reason = reason.as_str(),
+                        error = ?e,
+                        "Erasure decode stopped after downstream closed"
+                    );
+                } else {
+                    error!(
+                        block_offset,
+                        block_length,
+                        bytes_written = *written,
+                        stage = GET_STAGE_EMIT,
+                        reason = reason.as_str(),
+                        error = ?e,
+                        "Erasure decode failed to emit reconstructed data"
+                    );
+                }
                 *ret_err = Some(e);
                 return StripeFlow::Stop;
             }
@@ -1945,7 +1958,7 @@ mod tests {
     use std::io::Cursor;
     use std::pin::Pin;
     use std::sync::{
-        Arc,
+        Arc, Mutex,
         atomic::{AtomicUsize, Ordering},
     };
     use std::task::{Context, Poll};
@@ -2120,6 +2133,59 @@ mod tests {
         }
     }
 
+    struct DownstreamClosedWriter;
+
+    impl AsyncWrite for DownstreamClosedWriter {
+        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<io::Result<usize>> {
+            Poll::Ready(Err(crate::diagnostics::get::mark_get_object_downstream_closed(io::Error::new(
+                ErrorKind::BrokenPipe,
+                "injected downstream close",
+            ))))
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    #[derive(Clone, Default)]
+    struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
+
+    struct CapturedLogWriter(Arc<Mutex<Vec<u8>>>);
+
+    impl CapturedLogs {
+        fn contents(&self) -> String {
+            String::from_utf8(self.0.lock().expect("captured logs mutex should not be poisoned").clone())
+                .expect("captured logs should be valid UTF-8")
+        }
+    }
+
+    impl std::io::Write for CapturedLogWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0
+                .lock()
+                .expect("captured logs mutex should not be poisoned")
+                .extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs {
+        type Writer = CapturedLogWriter;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            CapturedLogWriter(Arc::clone(&self.0))
+        }
+    }
+
     #[test]
     fn parallel_reader_constructor_variants_preserve_read_cost_and_verification_flags() {
         let erasure = Erasure::new(2, 1, 64);
@@ -2215,6 +2281,47 @@ mod tests {
         assert_eq!(err.to_string(), "injected emit failure");
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn erasure_decode_logs_reconstructed_downstream_close_at_debug() {
+        let logs = CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::DEBUG)
+            .with_writer(logs.clone())
+            .with_ansi(false)
+            .without_time()
+            .finish();
+        let _guard = tracing::subscriber::set_default(subscriber);
+
+        let erasure = Erasure::new(2, 1, 64);
+        let data: Vec<u8> = (0..64).collect();
+        let shard_size = erasure.shard_size();
+        let encoded = erasure.encode_data(&data).expect("test data should encode");
+        let readers = vec![
+            None,
+            Some(BitrotReader::new(
+                Cursor::new(encoded[1].to_vec()),
+                shard_size,
+                HashAlgorithm::None,
+                false,
+            )),
+            Some(BitrotReader::new(
+                Cursor::new(encoded[2].to_vec()),
+                shard_size,
+                HashAlgorithm::None,
+                false,
+            )),
+        ];
+
+        let mut writer = DownstreamClosedWriter;
+        let (written, err) = erasure.decode(&mut writer, readers, 0, data.len(), data.len()).await;
+
+        assert_eq!(written, 0);
+        assert_eq!(err.expect("downstream close must still terminate the GET").kind(), ErrorKind::BrokenPipe);
+        let captured = logs.contents();
+        assert!(captured.contains("Erasure decode stopped after downstream closed"));
+        assert!(!captured.contains("Erasure decode failed to emit reconstructed data"));
+    }
+
     #[tokio::test]
     async fn test_erasure_decode_rejects_reader_count_and_range_overflow() {
         let erasure = Erasure::new(2, 1, 64);

crates/ecstore/src/layout/endpoints.rs

@@ -22,6 +22,8 @@ use rustfs_config::{
     ENV_STARTUP_TOPOLOGY_WAIT_MODE, ENV_STARTUP_TOPOLOGY_WAIT_TIMEOUT, ENV_UNSAFE_BYPASS_DISK_CHECK,
 };
 use rustfs_utils::{XHost, check_local_server_addr, get_env_opt_str, get_host_ip, is_local_host};
+#[cfg(test)]
+use std::sync::{LazyLock, Mutex};
 use std::{
     collections::{BTreeMap, BTreeSet, HashMap, HashSet, hash_map::Entry},
     future::Future,
@@ -598,6 +600,58 @@ const DNS_RETRY_JITTER_PERCENT: u64 = 20;
 /// wait does not flood the log with one line per backoff tick.
 const TOPOLOGY_WARN_THROTTLE: Duration = Duration::from_secs(30);
 
+#[cfg(test)]
+static FORCED_LOCAL_HOST_RESOLUTION_TIMEOUTS: LazyLock<Mutex<HashSet<String>>> = LazyLock::new(|| Mutex::new(HashSet::new()));
+
+#[cfg(test)]
+struct LocalHostResolutionTimeoutGuard {
+    hosts: Vec<String>,
+}
+
+#[cfg(test)]
+impl Drop for LocalHostResolutionTimeoutGuard {
+    fn drop(&mut self) {
+        let mut forced_hosts = FORCED_LOCAL_HOST_RESOLUTION_TIMEOUTS
+            .lock()
+            .expect("local-host test resolver mutex poisoned");
+        for host in &self.hosts {
+            forced_hosts.remove(host);
+        }
+    }
+}
+
+#[cfg(test)]
+fn force_local_host_resolution_timeout_for_test(hosts: &[&str]) -> LocalHostResolutionTimeoutGuard {
+    let hosts = hosts.iter().map(|host| (*host).to_string()).collect::<Vec<_>>();
+    let mut forced_hosts = FORCED_LOCAL_HOST_RESOLUTION_TIMEOUTS
+        .lock()
+        .expect("local-host test resolver mutex poisoned");
+    forced_hosts.extend(hosts.iter().cloned());
+    LocalHostResolutionTimeoutGuard { hosts }
+}
+
+#[cfg(test)]
+fn local_host_resolution_timeout_forced(host: &Host<&str>) -> bool {
+    let host = match host {
+        Host::Domain(domain) => (*domain).to_string(),
+        Host::Ipv4(ip) => ip.to_string(),
+        Host::Ipv6(ip) => ip.to_string(),
+    };
+    FORCED_LOCAL_HOST_RESOLUTION_TIMEOUTS
+        .lock()
+        .expect("local-host test resolver mutex poisoned")
+        .contains(&host)
+}
+
+fn endpoint_is_local_host(host: Host<&str>, port: u16, local_port: u16) -> Result<bool> {
+    #[cfg(test)]
+    if local_host_resolution_timeout_forced(&host) {
+        return Err(Error::new(ErrorKind::TimedOut, "resolver timeout"));
+    }
+
+    is_local_host(host, port, local_port)
+}
+
 struct DnsRetryDeadline {
     started: Instant,
     timeout: Duration,
@@ -694,7 +748,7 @@ async fn resolve_local_host_with_retry(
     retry_dns_operation(
         || {
             let host = host.clone();
-            async move { is_local_host(host, port, local_port) }
+            async move { endpoint_is_local_host(host, port, local_port) }
         },
         async_sleep,
         dns_retry_deadline,
@@ -2231,6 +2285,9 @@ mod test {
     #[serial]
     #[tokio::test]
     async fn create_server_endpoints_bounds_kubernetes_alias_dns_fallback() {
+        let _resolution_timeout =
+            force_local_host_resolution_timeout_for_test(&["unrelated-0.example.invalid", "unrelated-1.example.invalid"]);
+
         async_with_vars(
             [
                 (ENV_LOCAL_ENDPOINT_HOST, None),

crates/ecstore/src/lib.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 /// Scope-based hotpath measurement for `#[async_trait]` methods, where
 /// `#[hotpath::measure]` would only time the boxed-future construction.
 /// The guard records wall time from this statement until the enclosing

crates/ecstore/src/store/list_objects.rs

@@ -4861,10 +4861,7 @@ async fn poll_merge_head(rx: &CancellationToken, in_channels: &mut [Receiver<Met
 
 async fn send_or_cancel(rx: &CancellationToken, out_channel: &Sender<MetaCacheEntry>, entry: MetaCacheEntry) -> Result<bool> {
     tokio::select! {
-        result = out_channel.send(entry) => {
-            result.map_err(Error::other)?;
-            Ok(true)
-        }
+        result = out_channel.send(entry) => Ok(result.is_ok()),
         _ = rx.cancelled() => Ok(false),
     }
 }
@@ -9858,6 +9855,28 @@ mod test {
         assert_eq!(results, vec!["obj-a", "obj-b"]);
     }
 
+    #[tokio::test]
+    async fn merge_entry_channels_treats_dropped_output_receiver_as_completion() {
+        let (tx_a, rx_a) = mpsc::channel(4);
+        let (tx_b, rx_b) = mpsc::channel(4);
+        let (out_tx, out_rx) = mpsc::channel(1);
+
+        tx_a.send(test_meta_entry("obj-a")).await.unwrap();
+        tx_b.send(test_meta_entry("obj-b")).await.unwrap();
+        drop(tx_a);
+        drop(tx_b);
+        drop(out_rx);
+
+        let result = timeout(
+            Duration::from_secs(1),
+            merge_entry_channels(CancellationToken::new(), vec![rx_a, rx_b], out_tx, 1),
+        )
+        .await
+        .expect("merge should stop promptly when its consumer disconnects");
+
+        assert!(result.is_ok(), "consumer disconnect must not surface as a merge worker error");
+    }
+
     #[test]
     fn walk_ascending_versions_contract_reverses_newest_first_metadata() {
         // Documents the invariant the walk `versions_sort` handling relies on:

crates/extension-schema/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["web-programming", "development-tools"]
 [lib]
 doctest = false
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 serde = { workspace = true, features = ["derive"] }
 thiserror.workspace = true
 

crates/filemeta/Cargo.toml

@@ -27,9 +27,9 @@ documentation = "https://docs.rs/rustfs-filemeta/latest/rustfs_filemeta/"
 
 [features]
 default = []
-hotpath = ["hotpath/hotpath", "hotpath/tokio"]
-hotpath-alloc = ["hotpath/hotpath-alloc"]
-hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-utils/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
 
 [dependencies]
 hotpath.workspace = true

crates/heal/Cargo.toml

@@ -29,7 +29,48 @@ categories = ["web-programming", "development-tools", "filesystem"]
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "rustfs-common/hotpath",
+    "rustfs-concurrency/hotpath",
+    "rustfs-config/hotpath",
+    "rustfs-ecstore/hotpath",
+    "rustfs-madmin/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-utils/hotpath",
+    "rustfs-test-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-concurrency/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-madmin/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+    "rustfs-test-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-concurrency/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-ecstore/hotpath-cpu",
+    "rustfs-madmin/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+    "rustfs-test-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-config = { workspace = true }
 rustfs-concurrency = { workspace = true }
 rustfs-ecstore = { workspace = true }

crates/heal/src/heal/erasure_healer.rs

@@ -159,6 +159,7 @@ impl ErasureSetHealer {
 
     /// execute erasure set heal with resume
     #[tracing::instrument(skip(self, buckets), fields(set_disk_id = %set_disk_id, bucket_count = buckets.len()))]
+    #[hotpath::measure]
     pub async fn heal_erasure_set(&self, buckets: &[String], set_disk_id: &str) -> Result<()> {
         debug!(
             target: "rustfs::heal::erasure_healer",

crates/heal/src/heal/task.rs

@@ -584,6 +584,7 @@ impl HealTask {
     }
 
     #[tracing::instrument(skip(self), fields(task_id = %self.id, heal_type = ?self.heal_type))]
+    #[hotpath::measure]
     pub async fn execute(&self) -> Result<()> {
         // update status and timestamps atomically to avoid race conditions
         let now = SystemTime::now();
@@ -759,6 +760,7 @@ impl HealTask {
 
     // specific heal implementation method
     #[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))]
+    #[hotpath::measure]
     async fn heal_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
         debug!(
             target: "rustfs::heal::task",
@@ -1404,6 +1406,7 @@ impl HealTask {
         self.heal_bucket_objects(bucket, prefix).await
     }
 
+    #[hotpath::measure]
     async fn heal_bucket_objects(&self, bucket: &str, prefix: &str) -> Result<()> {
         let mut continuation_token: Option<String> = None;
         let mut scanned = 0u64;

crates/iam/Cargo.toml

@@ -28,7 +28,55 @@ documentation = "https://docs.rs/rustfs-iam/latest/rustfs_iam/"
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/reqwest-0-13",
+    "rustfs-config/hotpath",
+    "rustfs-credentials/hotpath",
+    "rustfs-crypto/hotpath",
+    "rustfs-ecstore/hotpath",
+    "rustfs-io-metrics/hotpath",
+    "rustfs-madmin/hotpath",
+    "rustfs-policy/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-utils/hotpath",
+    "rustfs-test-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-credentials/hotpath-alloc",
+    "rustfs-crypto/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-io-metrics/hotpath-alloc",
+    "rustfs-madmin/hotpath-alloc",
+    "rustfs-policy/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+    "rustfs-test-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-credentials/hotpath-cpu",
+    "rustfs-crypto/hotpath-cpu",
+    "rustfs-ecstore/hotpath-cpu",
+    "rustfs-io-metrics/hotpath-cpu",
+    "rustfs-madmin/hotpath-cpu",
+    "rustfs-policy/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+    "rustfs-test-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-credentials = { workspace = true }
 rustfs-config = { workspace = true, features = ["server-config-model"] }
 tokio = { workspace = true, features = ["fs", "rt-multi-thread"] }

crates/iam/src/store/object.rs

@@ -469,6 +469,7 @@ impl ObjectStore {
         });
     }
 
+    #[hotpath::measure]
     async fn list_all_iamconfig_items(&self) -> Result<HashMap<String, Vec<String>>> {
         let (tx, mut rx) = mpsc::channel::<StringOrErr>(100);
 
@@ -508,6 +509,7 @@ impl ObjectStore {
         Ok(res)
     }
 
+    #[hotpath::measure]
     async fn load_policy_doc_concurrent(&self, names: &[String], mode: LoadMode) -> Result<Vec<PolicyDoc>> {
         let mut futures = Vec::with_capacity(names.len());
 

crates/io-core/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["development-tools", "filesystem"]
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-io-metrics/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-io-metrics/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-io-metrics/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 bytes = { workspace = true, features = ["serde"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["io-util", "fs", "sync", "rt-multi-thread"] }

crates/io-metrics/Cargo.toml

@@ -28,7 +28,32 @@ categories = ["development-tools", "filesystem"]
 name = "metrics_pipeline"
 harness = false
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "rustfs-common/hotpath",
+    "rustfs-s3-ops/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-s3-ops/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-s3-ops/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 metrics = { workspace = true }
 rustfs-common = { workspace = true }
 rustfs-s3-ops = { workspace = true }

crates/keystone/Cargo.toml

@@ -28,7 +28,34 @@ authors.workspace = true
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/reqwest-0-13",
+    "rustfs-credentials/hotpath",
+    "rustfs-policy/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-credentials/hotpath-alloc",
+    "rustfs-policy/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-credentials/hotpath-cpu",
+    "rustfs-policy/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 tokio = { workspace = true, features = ["rt", "sync"] }
 reqwest = { workspace = true, features = ["json"] }
 serde = { workspace = true, features = ["derive"] }

crates/keystone/src/client.rs

@@ -95,6 +95,7 @@ impl KeystoneClient {
     }
 
     /// Validate a Keystone token
+    #[hotpath::measure]
     pub async fn validate_token(&self, token: &str) -> Result<KeystoneToken> {
         match self.version {
             KeystoneVersion::V3 => self.validate_token_v3(token).await,
@@ -238,6 +239,7 @@ impl KeystoneClient {
     }
 
     /// Get EC2 credentials for a user
+    #[hotpath::measure]
     pub async fn get_ec2_credentials(&self, user_id: &str, project_id: Option<&str>) -> Result<Vec<EC2Credential>> {
         let admin_token = self.get_admin_token().await?;
 

crates/kms/Cargo.toml

@@ -28,6 +28,7 @@ categories = ["cryptography", "web-programming", "authentication"]
 workspace = true
 
 [dependencies]
+hotpath.workspace = true
 # Core dependencies
 async-trait = { workspace = true }
 tokio = { workspace = true, features = ["fs", "io-util", "macros", "rt-multi-thread", "sync", "time"] }
@@ -84,3 +85,17 @@ tokio = { workspace = true, features = ["net", "test-util"] }
 
 [features]
 default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/reqwest-0-13",
+    "rustfs-security-governance/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-security-governance/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-security-governance/hotpath-cpu", "rustfs-utils/hotpath-cpu"]

crates/kms/src/backends/contract_tests.rs

@@ -27,11 +27,11 @@
 //! server, so they are `#[ignore]`d in CI. Static is covered by its own
 //! stateless contract below.
 
+use super::KmsBackend;
 use super::local::LocalKmsBackend;
 use super::static_kms::StaticKmsBackend;
 use super::vault::VaultKmsBackend;
 use super::vault_transit::VaultTransitKmsBackend;
-use super::{KmsBackend, KmsClient};
 use crate::config::KmsConfig;
 use crate::error::{KmsError, Result};
 use crate::manager::KmsManager;
@@ -46,6 +46,24 @@ use rand::RngExt as _;
 use std::collections::HashMap;
 use std::sync::Arc;
 
+fn expect_unsupported<T: std::fmt::Debug>(result: Result<T>) {
+    match result {
+        Err(KmsError::UnsupportedCapability { .. }) => {}
+        other => panic!("expected UnsupportedCapability, got {other:?}"),
+    }
+}
+
+/// Rotation while not Enabled: backends with rotation support must reject it
+/// through the state machine; backends without it report the capability gap.
+async fn expect_rotate_rejected(backend: &dyn KmsBackend, key_id: &str) {
+    let result = backend.rotate_key(key_id).await;
+    if backend.capabilities().rotate {
+        expect_invalid_key_state(result, "");
+    } else {
+        expect_unsupported(result);
+    }
+}
+
 fn expect_invalid_key_state<T: std::fmt::Debug>(result: Result<T>, expected_fragment: &str) {
     match result {
         Err(KmsError::InvalidOperation { message }) => assert!(
@@ -117,11 +135,9 @@ async fn assert_key_state(backend: &dyn KmsBackend, key_id: &str, expected: KeyS
     assert_eq!(described.key_metadata.key_state, expected, "unexpected state for key {key_id}");
 }
 
-/// Drives one freshly created (Enabled) key through the full state matrix.
-///
-/// `backend` is the product surface; `client` drives the lifecycle
-/// transitions not yet exposed through `KmsBackend`.
-async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn KmsClient, key_id: &str) {
+/// Drives one freshly created (Enabled) key through the full state matrix,
+/// entirely through the `KmsBackend` product surface.
+async fn assert_state_machine_contract(backend: &dyn KmsBackend, key_id: &str) {
     // Enabled: cryptographic use is allowed. Keep an envelope around to prove
     // decryption keeps working in later states.
     let data_key = backend
@@ -134,16 +150,13 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn Km
         .expect("Enabled key must encrypt");
 
     // Enabled -> Disabled.
-    client
-        .disable_key(key_id, None)
-        .await
-        .expect("disable from Enabled must succeed");
+    backend.disable_key(key_id).await.expect("disable from Enabled must succeed");
     assert_key_state(backend, key_id, KeyState::Disabled).await;
 
     // Disabled: new cryptographic use and rotation are rejected...
     expect_invalid_key_state(backend.encrypt(encrypt_request(key_id)).await, "disabled");
     expect_invalid_key_state(backend.generate_data_key(generate_request(key_id)).await, "disabled");
-    expect_invalid_key_state(client.rotate_key(key_id, None).await, "");
+    expect_rotate_rejected(backend, key_id).await;
     // ...but decryption of existing data keeps working (explicit AWS deviation)...
     let decrypted = backend
         .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
@@ -151,17 +164,14 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn Km
         .expect("decrypt with a disabled key must keep working");
     assert_eq!(decrypted.plaintext, data_key.plaintext_key, "decrypt must recover the original data key");
     // ...disable stays idempotent, cancel has nothing to cancel, and enable recovers.
-    client.disable_key(key_id, None).await.expect("disable must be idempotent");
+    backend.disable_key(key_id).await.expect("disable must be idempotent");
     expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "not pending deletion");
-    client
-        .enable_key(key_id, None)
-        .await
-        .expect("enable from Disabled must succeed");
+    backend.enable_key(key_id).await.expect("enable from Disabled must succeed");
     assert_key_state(backend, key_id, KeyState::Enabled).await;
 
     // Disabled keys may still be scheduled for deletion.
-    client
-        .disable_key(key_id, None)
+    backend
+        .disable_key(key_id)
         .await
         .expect("disable before scheduling must succeed");
     backend
@@ -173,10 +183,9 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, client: &dyn Km
     // PendingDeletion: everything except decryption and cancellation is rejected.
     expect_invalid_key_state(backend.encrypt(encrypt_request(key_id)).await, "pending deletion");
     expect_invalid_key_state(backend.generate_data_key(generate_request(key_id)).await, "pending deletion");
-    expect_invalid_key_state(client.enable_key(key_id, None).await, "pending deletion");
-    expect_invalid_key_state(client.disable_key(key_id, None).await, "pending deletion");
-    expect_invalid_key_state(client.rotate_key(key_id, None).await, "");
-    expect_invalid_key_state(client.schedule_key_deletion(key_id, 7, None).await, "pending deletion");
+    expect_invalid_key_state(backend.enable_key(key_id).await, "pending deletion");
+    expect_invalid_key_state(backend.disable_key(key_id).await, "pending deletion");
+    expect_rotate_rejected(backend, key_id).await;
     expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "pending deletion");
     let decrypted = backend
         .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
@@ -215,7 +224,7 @@ async fn local_fixture() -> (tempfile::TempDir, KmsConfig, LocalKmsBackend, Stri
 #[tokio::test]
 async fn local_backend_state_machine_contract() {
     let (_temp_dir, _config, backend, key_id) = local_fixture().await;
-    assert_state_machine_contract(&backend, backend.lifecycle_client(), &key_id).await;
+    assert_state_machine_contract(&backend, &key_id).await;
 }
 
 /// SSE-shaped regression: disabling a key must not break decryption of data
@@ -256,10 +265,7 @@ async fn static_backend_stateless_contract() {
     rand::rng().fill(&mut raw_key[..]);
     let config = KmsConfig::static_kms(key_id.to_string(), BASE64.encode(raw_key));
     let static_backend = StaticKmsBackend::new(config).await.expect("static backend should build");
-    // StaticKmsBackend implements both traits with overlapping method names,
-    // so pin each surface once instead of qualifying every call.
     let backend: &dyn KmsBackend = &static_backend;
-    let client: &dyn KmsClient = &static_backend;
 
     let data_key = backend
         .generate_data_key(generate_request(key_id))
@@ -275,9 +281,11 @@ async fn static_backend_stateless_contract() {
     expect_invalid_key_state(backend.create_key(create_request("another-key".to_string())).await, "read-only");
     expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "read-only");
     expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "read-only");
-    expect_invalid_key_state(client.disable_key(key_id, None).await, "read-only");
-    expect_invalid_key_state(client.schedule_key_deletion(key_id, 7, None).await, "read-only");
-    expect_invalid_key_state(client.rotate_key(key_id, None).await, "read-only");
+    // Enable/disable and rotation are capability gaps at the product
+    // surface, not state-machine rejections.
+    expect_unsupported(backend.enable_key(key_id).await);
+    expect_unsupported(backend.disable_key(key_id).await);
+    expect_unsupported(backend.rotate_key(key_id).await);
 }
 
 fn vault_dev_config(constructor: fn(url::Url, String) -> KmsConfig) -> KmsConfig {
@@ -298,7 +306,15 @@ async fn vault_kv2_backend_state_machine_contract() {
         .await
         .expect("key should be created");
 
-    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
+    assert_state_machine_contract(&backend, &created.key_id).await;
+
+    // KV2 additionally supports version-retaining rotation, which must only
+    // work while the key is Enabled (the shared matrix covered the
+    // rejections).
+    backend
+        .rotate_key(&created.key_id)
+        .await
+        .expect("rotation of an Enabled KV2 key must succeed");
 
     // Cleanup: leave the key pending deletion so repeated runs stay tidy.
     let _ = backend.delete_key(schedule_request(&created.key_id)).await;
@@ -316,13 +332,12 @@ async fn vault_transit_backend_state_machine_contract() {
         .await
         .expect("key should be created");
 
-    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
+    assert_state_machine_contract(&backend, &created.key_id).await;
 
     // Transit additionally supports rotation, which must only work while the
     // key is Enabled (the shared matrix already covered the rejections).
     backend
-        .lifecycle_client()
-        .rotate_key(&created.key_id, None)
+        .rotate_key(&created.key_id)
         .await
         .expect("rotation of an Enabled transit key must succeed");
 

crates/kms/src/backends/local.rs

@@ -14,9 +14,7 @@
 
 //! Local file-based KMS backend implementation
 
-use crate::backends::{
-    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_status_permits,
-};
+use crate::backends::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_status_permits};
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -399,6 +397,20 @@ pub struct LocalKmsClient {
     /// Per-key write locks serializing read-modify-write updates within this
     /// process (see [`Self::lock_key_for_write`]).
     key_write_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
+    /// Directory-wide writer fence for backup export (see
+    /// [`Self::acquire_export_fence`]). Writers hold the read side; an export
+    /// snapshot holds the write side so it observes a single-generation view.
+    export_fence: Arc<tokio::sync::RwLock<()>>,
+}
+
+/// Guard pairing the export-fence read lock with a per-key write mutex.
+///
+/// Dropping it releases both, so every existing `lock_key_for_write` call
+/// site participates in the export fence without changes.
+#[must_use]
+struct KeyWriteGuard {
+    _fence: tokio::sync::OwnedRwLockReadGuard<()>,
+    _key: tokio::sync::OwnedMutexGuard<()>,
 }
 
 // pub(crate) so the backup contract tests can anchor the manifest's
@@ -465,6 +477,7 @@ impl LocalKmsClient {
             legacy_master_cipher,
             dek_crypto: AesDekCrypto::new(),
             key_write_locks: Mutex::new(HashMap::new()),
+            export_fence: Arc::new(tokio::sync::RwLock::new(())),
         };
         client.validate_existing_keys().await?;
         Ok(client)
@@ -507,6 +520,7 @@ impl LocalKmsClient {
             legacy_master_cipher,
             dek_crypto: AesDekCrypto::new(),
             key_write_locks: Mutex::new(HashMap::new()),
+            export_fence: Arc::new(tokio::sync::RwLock::new(())),
         })
     }
 
@@ -517,12 +531,40 @@ impl LocalKmsClient {
     /// delete with a rewrite. Cross-process writers sharing a key directory
     /// remain unsupported. Entries live for the client's lifetime; the table
     /// is bounded by the number of distinct key ids this process touches.
-    async fn lock_key_for_write(&self, key_id: &str) -> tokio::sync::OwnedMutexGuard<()> {
+    async fn lock_key_for_write(&self, key_id: &str) -> KeyWriteGuard {
+        // Fence first, per-key mutex second: the ordering is uniform across
+        // all writers, so an export waiting on the write side can never
+        // deadlock with a writer holding a key mutex.
+        let fence = Arc::clone(&self.export_fence).read_owned().await;
         let lock = {
             let mut locks = self.key_write_locks.lock().expect("Local KMS key write lock table poisoned");
             Arc::clone(locks.entry(key_id.to_string()).or_default())
         };
-        lock.lock_owned().await
+        KeyWriteGuard {
+            _fence: fence,
+            _key: lock.lock_owned().await,
+        }
+    }
+
+    /// Block every key-directory writer while a backup export collects its
+    /// snapshot, so all records belong to one generation.
+    ///
+    /// Mutating operations hold the read side (via [`Self::lock_key_for_write`]
+    /// or [`Self::save_new_master_key`]); the export holds the write side only
+    /// for the collection phase, never while encrypting or writing the bundle.
+    pub(crate) async fn acquire_export_fence(&self) -> tokio::sync::OwnedRwLockWriteGuard<()> {
+        Arc::clone(&self.export_fence).write_owned().await
+    }
+
+    /// Key directory root, exposed for the backup export module.
+    pub(crate) fn key_directory(&self) -> &Path {
+        &self.config.key_dir
+    }
+
+    /// Absolute path of the master-key KDF salt file, exposed for the backup
+    /// export module.
+    pub(crate) fn master_key_salt_file(&self) -> PathBuf {
+        Self::master_key_salt_path(&self.config)
     }
 
     /// Derive a 256-bit key from the master key string using a persistent Argon2id salt.
@@ -799,6 +841,11 @@ impl LocalKmsClient {
     }
 
     async fn save_new_master_key(&self, master_key: &MasterKeyInfo, key_material: &[u8]) -> Result<()> {
+        // Creates never take the per-key write lock (`NoClobber` publishing
+        // already linearizes them), so they join the export fence here. This
+        // must stay the only fence acquisition on the create path: the fence
+        // read lock is not reentrant while an export waits for the write side.
+        let _fence = Arc::clone(&self.export_fence).read_owned().await;
         let key_path = self.master_key_path(&master_key.key_id)?;
         let content = self.encode_master_key(master_key, key_material)?;
         let temp_path = key_path.with_extension(format!("tmp-{}", uuid::Uuid::new_v4()));
@@ -937,9 +984,12 @@ impl LocalKmsClient {
     }
 }
 
-#[async_trait]
-impl KmsClient for LocalKmsClient {
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+impl LocalKmsClient {
+    pub(crate) async fn generate_data_key(
+        &self,
+        request: &GenerateKeyRequest,
+        context: Option<&OperationContext>,
+    ) -> Result<DataKeyInfo> {
         debug!("Generating data key for master key: {}", request.master_key_id);
 
         let key_info = self.describe_key(&request.master_key_id, context).await?;
@@ -980,7 +1030,7 @@ impl KmsClient for LocalKmsClient {
         Ok(data_key)
     }
 
-    async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    pub(crate) async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse> {
         debug!("Encrypting data with key: {}", request.key_id);
 
         // Verify key exists and its state allows encryption
@@ -997,7 +1047,7 @@ impl KmsClient for LocalKmsClient {
         })
     }
 
-    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    pub(crate) async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
         debug!("Decrypting data");
 
         // Parse the data key envelope from ciphertext
@@ -1031,7 +1081,14 @@ impl KmsClient for LocalKmsClient {
         Ok(plaintext)
     }
 
-    async fn create_key(&self, key_id: &str, algorithm: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn create_key(
+        &self,
+        key_id: &str,
+        algorithm: &str,
+        context: Option<&OperationContext>,
+    ) -> Result<MasterKeyInfo> {
         debug!("Creating master key: {}", key_id);
 
         // Check if key already exists
@@ -1060,14 +1117,18 @@ impl KmsClient for LocalKmsClient {
         Ok(master_key)
     }
 
-    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
+    pub(crate) async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
         debug!("Describing key: {}", key_id);
 
         let master_key = self.load_master_key(key_id).await?;
         Ok(master_key.into())
     }
 
-    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
+    pub(crate) async fn list_keys(
+        &self,
+        request: &ListKeysRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<ListKeysResponse> {
         debug!("Listing keys");
 
         let mut keys = Vec::new();
@@ -1111,7 +1172,7 @@ impl KmsClient for LocalKmsClient {
         })
     }
 
-    async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Enabling key: {}", key_id);
 
         let _write_guard = self.lock_key_for_write(key_id).await;
@@ -1129,7 +1190,7 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Disabling key: {}", key_id);
 
         let _write_guard = self.lock_key_for_write(key_id).await;
@@ -1146,7 +1207,9 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn schedule_key_deletion(
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn schedule_key_deletion(
         &self,
         key_id: &str,
         pending_window_days: u32,
@@ -1170,7 +1233,9 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Canceling deletion for key: {}", key_id);
 
         let _write_guard = self.lock_key_for_write(key_id).await;
@@ -1190,7 +1255,9 @@ impl KmsClient for LocalKmsClient {
         Ok(())
     }
 
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         if !fs::try_exists(self.master_key_path(key_id)?).await? {
             return Err(KmsError::key_not_found(key_id));
         }
@@ -1199,7 +1266,7 @@ impl KmsClient for LocalKmsClient {
         ))
     }
 
-    async fn health_check(&self) -> Result<()> {
+    pub(crate) async fn health_check(&self) -> Result<()> {
         // Check if key directory is accessible
         if !self.config.key_dir.exists() {
             return Err(KmsError::backend_error("Key directory does not exist"));
@@ -1210,17 +1277,6 @@ impl KmsClient for LocalKmsClient {
 
         Ok(())
     }
-
-    fn backend_info(&self) -> BackendInfo {
-        BackendInfo::new(
-            "local".to_string(),
-            env!("CARGO_PKG_VERSION").to_string(),
-            self.config.key_dir.to_string_lossy().to_string(),
-            true, // We'll assume healthy for now
-        )
-        .with_metadata("key_dir".to_string(), self.config.key_dir.to_string_lossy().to_string())
-        .with_metadata("encrypted_at_rest".to_string(), self.master_cipher.is_some().to_string())
-    }
 }
 
 /// LocalKmsBackend wraps LocalKmsClient and implements the KmsBackend trait

crates/kms/src/backends/mod.rs

@@ -19,7 +19,6 @@ use crate::types::*;
 use async_trait::async_trait;
 use jiff::Zoned;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
 
 #[cfg(test)]
 mod contract_tests;
@@ -99,136 +98,6 @@ pub(crate) fn ensure_key_status_permits(key_id: &str, status: &KeyStatus, operat
     ensure_key_state_permits(key_id, &state, operation)
 }
 
-/// Abstract KMS client interface that all backends must implement
-#[async_trait]
-pub trait KmsClient: Send + Sync {
-    /// Generate a new data encryption key (DEK)
-    ///
-    /// Creates a new data key using the specified master key. The returned DataKey
-    /// contains both the plaintext and encrypted versions of the key.
-    ///
-    /// # Arguments
-    /// * `request` - The key generation request
-    /// * `context` - Optional operation context for auditing
-    ///
-    /// # Returns
-    /// Returns a DataKey containing both plaintext and encrypted key material
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, context: Option<&OperationContext>) -> Result<DataKeyInfo>;
-
-    /// Encrypt data directly using a master key
-    ///
-    /// Encrypts the provided plaintext using the specified master key.
-    /// This is different from generate_data_key as it encrypts user data directly.
-    ///
-    /// # Arguments
-    /// * `request` - The encryption request containing plaintext and key ID
-    /// * `context` - Optional operation context for auditing
-    async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse>;
-
-    /// Decrypt data using a master key
-    ///
-    /// Decrypts the provided ciphertext. The KMS automatically determines
-    /// which key was used for encryption based on the ciphertext metadata.
-    ///
-    /// # Arguments
-    /// * `request` - The decryption request containing ciphertext
-    /// * `context` - Optional operation context for auditing
-    async fn decrypt(&self, request: &DecryptRequest, context: Option<&OperationContext>) -> Result<Vec<u8>>;
-
-    /// Create a new master key
-    ///
-    /// Creates a new master key in the KMS with the specified ID.
-    /// Returns an error if a key with the same ID already exists.
-    ///
-    /// # Arguments
-    /// * `key_id` - Unique identifier for the new key
-    /// * `algorithm` - Key algorithm (e.g., "AES_256")
-    /// * `context` - Optional operation context for auditing
-    async fn create_key(&self, key_id: &str, algorithm: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo>;
-
-    /// Get information about a specific key
-    ///
-    /// Returns metadata and information about the specified key.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn describe_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<KeyInfo>;
-
-    /// List available keys
-    ///
-    /// Returns a paginated list of keys available in the KMS.
-    ///
-    /// # Arguments
-    /// * `request` - List request parameters (pagination, filters)
-    /// * `context` - Optional operation context for auditing
-    async fn list_keys(&self, request: &ListKeysRequest, context: Option<&OperationContext>) -> Result<ListKeysResponse>;
-
-    /// Enable a key
-    ///
-    /// Enables a previously disabled key, allowing it to be used for cryptographic operations.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn enable_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<()>;
-
-    /// Disable a key
-    ///
-    /// Disables a key, preventing it from being used for new cryptographic operations.
-    /// Existing encrypted data can still be decrypted.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn disable_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<()>;
-
-    /// Schedule key deletion
-    ///
-    /// Schedules a key for deletion after a specified number of days.
-    /// This allows for a grace period to recover the key if needed.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `pending_window_days` - Number of days before actual deletion
-    /// * `context` - Optional operation context for auditing
-    async fn schedule_key_deletion(
-        &self,
-        key_id: &str,
-        pending_window_days: u32,
-        context: Option<&OperationContext>,
-    ) -> Result<()>;
-
-    /// Cancel key deletion
-    ///
-    /// Cancels a previously scheduled key deletion.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn cancel_key_deletion(&self, key_id: &str, context: Option<&OperationContext>) -> Result<()>;
-
-    /// Rotate a key
-    ///
-    /// Creates a new version of the specified key. Previous versions remain
-    /// available for decryption but new operations will use the new version.
-    ///
-    /// # Arguments
-    /// * `key_id` - The key identifier
-    /// * `context` - Optional operation context for auditing
-    async fn rotate_key(&self, key_id: &str, context: Option<&OperationContext>) -> Result<MasterKeyInfo>;
-
-    /// Health check
-    ///
-    /// Performs a health check on the KMS backend to ensure it's operational.
-    async fn health_check(&self) -> Result<()>;
-
-    /// Get backend information
-    ///
-    /// Returns information about the KMS backend (type, version, etc.).
-    fn backend_info(&self) -> BackendInfo;
-}
-
 /// Simplified KMS backend interface for manager
 #[async_trait]
 pub trait KmsBackend: Send + Sync {
@@ -327,58 +196,6 @@ pub enum ExpiredKeyRemoval {
     NotExpired,
 }
 
-/// Information about a KMS backend
-#[derive(Debug, Clone)]
-pub struct BackendInfo {
-    /// Backend type name (e.g., "local", "vault")
-    pub backend_type: String,
-    /// Backend version
-    pub version: String,
-    /// Backend endpoint or location
-    pub endpoint: String,
-    /// Whether the backend is currently healthy
-    pub healthy: bool,
-    /// Additional metadata about the backend
-    pub metadata: HashMap<String, String>,
-}
-
-impl BackendInfo {
-    /// Create a new backend info
-    ///
-    /// # Arguments
-    /// * `backend_type` - The type of the backend
-    /// * `version` - The version of the backend
-    /// * `endpoint` - The endpoint or location of the backend
-    /// * `healthy` - Whether the backend is healthy
-    ///
-    /// # Returns
-    /// A new BackendInfo instance
-    ///
-    pub fn new(backend_type: String, version: String, endpoint: String, healthy: bool) -> Self {
-        Self {
-            backend_type,
-            version,
-            endpoint,
-            healthy,
-            metadata: HashMap::new(),
-        }
-    }
-
-    /// Add metadata to the backend info
-    ///
-    /// # Arguments
-    /// * `key` - Metadata key
-    /// * `value` - Metadata value
-    ///
-    /// # Returns
-    /// Updated BackendInfo instance
-    ///
-    pub fn with_metadata(mut self, key: String, value: String) -> Self {
-        self.metadata.insert(key, value);
-        self
-    }
-}
-
 /// Set of operations a KMS backend supports.
 ///
 /// Reported by [`KmsBackend::capabilities`] so callers (manager, admin API)

crates/kms/src/backends/scripted_vault.rs

@@ -26,16 +26,16 @@ use std::sync::{Arc, Mutex};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
 
-/// One canned HTTP response.
-pub(crate) struct ScriptedResponse {
-    status: u16,
-    body: String,
+/// One scripted connection outcome.
+pub(crate) enum ScriptedResponse {
+    Http { status: u16, body: String },
+    Close,
 }
 
 impl ScriptedResponse {
     /// A 200 response carrying `data` inside the standard Vault envelope.
     pub(crate) fn ok(data: serde_json::Value) -> Self {
-        Self {
+        Self::Http {
             status: 200,
             body: serde_json::json!({
                 "request_id": "scripted",
@@ -50,11 +50,16 @@ impl ScriptedResponse {
 
     /// An error response in Vault's `{"errors": [...]}` format.
     pub(crate) fn error(status: u16, message: &str) -> Self {
-        Self {
+        Self::Http {
             status,
             body: serde_json::json!({ "errors": [message] }).to_string(),
         }
     }
+
+    /// Close the connection after consuming a request without sending an HTTP response.
+    pub(crate) fn close() -> Self {
+        Self::Close
+    }
 }
 
 /// A scripted stand-in Vault listening on a loopback port.
@@ -91,14 +96,14 @@ impl ScriptedVault {
                 let response = responses
                     .next()
                     .unwrap_or_else(|| ScriptedResponse::error(599, "scripted vault: script exhausted"));
-                let payload = format!(
-                    "HTTP/1.1 {} Scripted\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
-                    response.status,
-                    response.body.len(),
-                    response.body
-                );
-                let _ = stream.write_all(payload.as_bytes()).await;
-                let _ = stream.shutdown().await;
+                if let ScriptedResponse::Http { status, body } = response {
+                    let payload = format!(
+                        "HTTP/1.1 {status} Scripted\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
+                        body.len(),
+                    );
+                    let _ = stream.write_all(payload.as_bytes()).await;
+                    let _ = stream.shutdown().await;
+                }
             }
         });
 

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_kv2_backend_capabilities.snap

@@ -8,7 +8,7 @@ expression: capabilities_snapshot(backend.capabilities())
   "encrypt": true,
   "generate_data_key": true,
   "physical_delete": true,
-  "rotate": false,
+  "rotate": true,
   "schedule_deletion": true,
-  "versioning": false
+  "versioning": true
 }

crates/kms/src/backends/static_kms.rs

@@ -21,7 +21,7 @@
 //!
 //! encrypted_data(plaintext_len+16) || nonce (12 bytes)
 
-use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
+use crate::backends::{BackendCapabilities, KmsBackend};
 use crate::config::{BackendConfig, KmsConfig};
 use crate::encryption::DataKeyEnvelope;
 use crate::error::{KmsError, Result};
@@ -98,9 +98,10 @@ impl StaticKmsBackend {
     }
 }
 
-#[async_trait]
-impl KmsClient for StaticKmsBackend {
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+impl StaticKmsBackend {
+    /// Generate a fresh data key and wrap it in the standard KMS envelope,
+    /// authenticated against the canonical encryption context.
+    pub(crate) fn generate_data_key_envelope(&self, request: &GenerateKeyRequest) -> Result<DataKeyInfo> {
         if request.master_key_id != self.key_id {
             return Err(KmsError::key_not_found(&request.master_key_id));
         }
@@ -151,7 +152,8 @@ impl KmsClient for StaticKmsBackend {
         ))
     }
 
-    async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    /// Encrypt caller-provided plaintext into the standard KMS envelope.
+    pub(crate) fn encrypt_to_envelope(&self, request: &EncryptRequest) -> Result<EncryptResponse> {
         if request.key_id != self.key_id {
             return Err(KmsError::key_not_found(&request.key_id));
         }
@@ -196,7 +198,8 @@ impl KmsClient for StaticKmsBackend {
         })
     }
 
-    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    /// Open a KMS envelope produced by this backend.
+    pub(crate) fn decrypt_envelope(&self, request: &DecryptRequest) -> Result<Vec<u8>> {
         let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
             .map_err(|error| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {error}")))?;
         if envelope.master_key_id != self.key_id {
@@ -235,14 +238,8 @@ impl KmsClient for StaticKmsBackend {
         Ok(plaintext)
     }
 
-    async fn create_key(&self, key_id: &str, _algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
-        if key_id == self.key_id {
-            return Err(KmsError::key_already_exists(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot create new keys"))
-    }
-
-    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
+    /// Describe the single configured key.
+    pub(crate) fn configured_key_info(&self, key_id: &str) -> Result<KeyInfo> {
         if key_id != self.key_id {
             return Err(KmsError::key_not_found(key_id));
         }
@@ -263,7 +260,8 @@ impl KmsClient for StaticKmsBackend {
         })
     }
 
-    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
+    /// List the single configured key, honouring the pagination marker.
+    pub(crate) fn list_configured_key(&self, request: &ListKeysRequest) -> Result<ListKeysResponse> {
         let key_info = KeyInfo {
             key_id: self.key_id.clone(),
             description: Some("Static single-key KMS backend".to_string()),
@@ -295,57 +293,6 @@ impl KmsClient for StaticKmsBackend {
             truncated: false,
         })
     }
-
-    async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        // Static KMS key is always enabled
-        Ok(())
-    }
-
-    async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot disable keys"))
-    }
-
-    async fn schedule_key_deletion(
-        &self,
-        key_id: &str,
-        _pending_window_days: u32,
-        _context: Option<&OperationContext>,
-    ) -> Result<()> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot schedule key deletion"))
-    }
-
-    async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot cancel key deletion"))
-    }
-
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
-        if key_id != self.key_id {
-            return Err(KmsError::key_not_found(key_id));
-        }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot rotate keys"))
-    }
-
-    async fn health_check(&self) -> Result<()> {
-        // Static KMS is always healthy if it was successfully initialized
-        Ok(())
-    }
-
-    fn backend_info(&self) -> BackendInfo {
-        BackendInfo::new("static".to_string(), env!("CARGO_PKG_VERSION").to_string(), "local".to_string(), true)
-            .with_metadata("key_id".to_string(), self.key_id.clone())
-    }
 }
 
 #[async_trait]
@@ -359,12 +306,12 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn encrypt(&self, request: EncryptRequest) -> Result<EncryptResponse> {
-        <Self as KmsClient>::encrypt(self, &request, None).await
+        self.encrypt_to_envelope(&request)
     }
 
     async fn decrypt(&self, request: DecryptRequest) -> Result<DecryptResponse> {
         let key_id = self.key_id.clone();
-        let plaintext = <Self as KmsClient>::decrypt(self, &request, None).await?;
+        let plaintext = self.decrypt_envelope(&request)?;
         Ok(DecryptResponse {
             plaintext,
             key_id,
@@ -380,7 +327,7 @@ impl KmsBackend for StaticKmsBackend {
             encryption_context: request.encryption_context,
             grant_tokens: Vec::new(),
         };
-        let data_key = <Self as KmsClient>::generate_data_key(self, &gen_req, None).await?;
+        let data_key = self.generate_data_key_envelope(&gen_req)?;
 
         let plaintext_key = data_key
             .plaintext
@@ -395,7 +342,7 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn describe_key(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
-        let key_info = <Self as KmsClient>::describe_key(self, &request.key_id, None).await?;
+        let key_info = self.configured_key_info(&request.key_id)?;
         let key_metadata = KeyMetadata {
             key_id: key_info.key_id.clone(),
             key_state: if key_info.status == KeyStatus::Active {
@@ -415,7 +362,7 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn list_keys(&self, request: ListKeysRequest) -> Result<ListKeysResponse> {
-        <Self as KmsClient>::list_keys(self, &request, None).await
+        self.list_configured_key(&request)
     }
 
     async fn delete_key(&self, request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
@@ -446,7 +393,7 @@ impl KmsBackend for StaticKmsBackend {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::backends::{KmsBackend as KmsBackendTrait, KmsClient};
+    use crate::backends::KmsBackend as KmsBackendTrait;
     use crate::config::{BackendConfig, KmsBackend, StaticConfig};
     use crate::encryption::is_data_key_envelope;
     use base64::Engine as _;
@@ -491,8 +438,8 @@ mod tests {
         // Generate data key
         let request = GenerateKeyRequest::new(key_id.clone(), "AES_256".to_string())
             .with_context("bucket".to_string(), "test-bucket".to_string());
-        let data_key = KmsClient::generate_data_key(&backend, &request, None)
-            .await
+        let data_key = backend
+            .generate_data_key_envelope(&request)
             .expect("Failed to generate data key");
 
         assert_eq!(data_key.key_id, key_id);
@@ -509,9 +456,7 @@ mod tests {
         // Decrypt the data key
         let decrypt_request =
             DecryptRequest::new(data_key.ciphertext.clone()).with_context("bucket".to_string(), "test-bucket".to_string());
-        let decrypted = KmsClient::decrypt(&backend, &decrypt_request, None)
-            .await
-            .expect("Failed to decrypt");
+        let decrypted = backend.decrypt_envelope(&decrypt_request).expect("Failed to decrypt");
 
         assert_eq!(decrypted.as_slice(), data_key.plaintext.as_deref().expect("plaintext should exist"));
     }
@@ -523,8 +468,8 @@ mod tests {
             .with_context("bucket".to_string(), "source-bucket".to_string())
             .with_context("object".to_string(), "source-object".to_string());
 
-        let data_key = KmsClient::generate_data_key(&backend, &request, None)
-            .await
+        let data_key = backend
+            .generate_data_key_envelope(&request)
             .expect("generate static KMS data key");
 
         assert!(
@@ -568,8 +513,8 @@ mod tests {
         let (backend, key_id, _key) = create_test_backend().await;
         let request = GenerateKeyRequest::new(key_id, "AES_256".to_string())
             .with_context("bucket".to_string(), "source-bucket".to_string());
-        let generated = KmsClient::generate_data_key(&backend, &request, None)
-            .await
+        let generated = backend
+            .generate_data_key_envelope(&request)
             .expect("generate context-bound data key");
         let mut envelope: DataKeyEnvelope = serde_json::from_slice(&generated.ciphertext).expect("parse static KMS envelope");
         envelope
@@ -578,8 +523,8 @@ mod tests {
         let decrypt_request = DecryptRequest::new(serde_json::to_vec(&envelope).expect("serialize tampered envelope"))
             .with_context("bucket".to_string(), "different-bucket".to_string());
 
-        let error = KmsClient::decrypt(&backend, &decrypt_request, None)
-            .await
+        let error = backend
+            .decrypt_envelope(&decrypt_request)
             .expect_err("tampering with authenticated envelope context must fail");
 
         assert!(matches!(error, KmsError::CryptographicError { .. }));
@@ -590,7 +535,7 @@ mod tests {
         let (backend, _key_id, _key) = create_test_backend().await;
 
         let request = GenerateKeyRequest::new("wrong-key-id".to_string(), "AES_256".to_string());
-        let result = KmsClient::generate_data_key(&backend, &request, None).await;
+        let result = backend.generate_data_key_envelope(&request);
         assert!(result.is_err());
         assert!(result.expect_err("should be Err").to_string().contains("wrong-key-id"));
     }
@@ -602,7 +547,7 @@ mod tests {
         // Ciphertext too short
         let short = vec![0u8; 10];
         let request = DecryptRequest::new(short);
-        let result = KmsClient::decrypt(&backend, &request, None).await;
+        let result = backend.decrypt_envelope(&request);
         assert!(result.is_err());
     }
 
@@ -612,9 +557,7 @@ mod tests {
 
         // Generate a valid ciphertext first
         let gen_request = GenerateKeyRequest::new(key_id, "AES_256".to_string());
-        let data_key = KmsClient::generate_data_key(&backend, &gen_request, None)
-            .await
-            .expect("generate");
+        let data_key = backend.generate_data_key_envelope(&gen_request).expect("generate");
 
         // Tamper with the ciphertext (flip a bit in the encrypted portion)
         let mut tampered = data_key.ciphertext.clone();
@@ -623,7 +566,7 @@ mod tests {
         }
 
         let request = DecryptRequest::new(tampered);
-        let result = KmsClient::decrypt(&backend, &request, None).await;
+        let result = backend.decrypt_envelope(&request);
         assert!(result.is_err());
     }
 
@@ -632,7 +575,14 @@ mod tests {
         let (backend, key_id, _key) = create_test_backend().await;
 
         // Creating the pre-configured key should return KeyAlreadyExists
-        let result = KmsClient::create_key(&backend, &key_id, "AES_256", None).await;
+        let result = KmsBackendTrait::create_key(
+            &backend,
+            CreateKeyRequest {
+                key_name: Some(key_id.clone()),
+                ..Default::default()
+            },
+        )
+        .await;
         assert!(result.is_err());
         assert!(result.expect_err("should be Err").to_string().contains("already exists"));
     }
@@ -642,7 +592,14 @@ mod tests {
         let (backend, _key_id, _key) = create_test_backend().await;
 
         // Creating any other key should return invalid operation (read-only)
-        let result = KmsClient::create_key(&backend, "other-key", "AES_256", None).await;
+        let result = KmsBackendTrait::create_key(
+            &backend,
+            CreateKeyRequest {
+                key_name: Some("other-key".to_string()),
+                ..Default::default()
+            },
+        )
+        .await;
         assert!(result.is_err());
         let err_msg = result.expect_err("should be Err").to_string();
         assert!(err_msg.contains("read-only") || err_msg.contains("cannot create"));
@@ -652,15 +609,13 @@ mod tests {
     async fn test_describe_key() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let key_info = KmsClient::describe_key(&backend, &key_id, None)
-            .await
-            .expect("describe_key should succeed");
+        let key_info = backend.configured_key_info(&key_id).expect("describe_key should succeed");
         assert_eq!(key_info.key_id, key_id);
         assert_eq!(key_info.status, KeyStatus::Active);
         assert_eq!(key_info.algorithm, "AES_256");
 
         // Wrong key ID
-        let result = KmsClient::describe_key(&backend, "nonexistent", None).await;
+        let result = backend.configured_key_info("nonexistent");
         assert!(result.is_err());
     }
 
@@ -668,8 +623,8 @@ mod tests {
     async fn test_list_keys() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let response = KmsClient::list_keys(&backend, &ListKeysRequest::default(), None)
-            .await
+        let response = backend
+            .list_configured_key(&ListKeysRequest::default())
             .expect("list_keys should succeed");
         assert_eq!(response.keys.len(), 1);
         assert_eq!(response.keys[0].key_id, key_id);
@@ -677,61 +632,45 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_disable_key_returns_error() {
+    async fn lifecycle_mutations_are_unsupported_at_the_product_surface() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let result = KmsClient::disable_key(&backend, &key_id, None).await;
-        assert!(result.is_err());
-        assert!(result.expect_err("should be Err").to_string().contains("read-only"));
-    }
-
-    #[tokio::test]
-    async fn test_enable_key_is_noop() {
-        let (backend, key_id, _key) = create_test_backend().await;
-
-        // Enable should succeed (no-op for static KMS)
-        KmsClient::enable_key(&backend, &key_id, None)
-            .await
-            .expect("enable_key should be no-op");
-
-        // Wrong key should still fail
-        let result = KmsClient::enable_key(&backend, "wrong", None).await;
-        assert!(result.is_err());
+        // The static backend advertises no enable/disable or rotation
+        // capability, so the shared KmsBackend defaults reject all three.
+        for result in [
+            KmsBackendTrait::enable_key(&backend, &key_id).await,
+            KmsBackendTrait::disable_key(&backend, &key_id).await,
+            KmsBackendTrait::rotate_key(&backend, &key_id).await,
+        ] {
+            let error = result.expect_err("static lifecycle mutations must be rejected");
+            assert!(matches!(error, KmsError::UnsupportedCapability { .. }), "got {error:?}");
+        }
     }
 
     #[tokio::test]
     async fn test_delete_key_returns_error() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let result = KmsClient::schedule_key_deletion(&backend, &key_id, 7, None).await;
+        let result = KmsBackendTrait::delete_key(
+            &backend,
+            DeleteKeyRequest {
+                key_id: key_id.clone(),
+                pending_window_in_days: Some(7),
+                force_immediate: None,
+            },
+        )
+        .await;
         assert!(result.is_err());
         assert!(result.expect_err("should be Err").to_string().contains("read-only"));
     }
 
-    #[tokio::test]
-    async fn test_rotate_key_returns_error() {
-        let (backend, key_id, _key) = create_test_backend().await;
-
-        let result = KmsClient::rotate_key(&backend, &key_id, None).await;
-        assert!(result.is_err());
-    }
-
     #[tokio::test]
     async fn test_health_check() {
         let (backend, _key_id, _key) = create_test_backend().await;
 
-        KmsClient::health_check(&backend).await.expect("health_check should succeed");
-    }
-
-    #[tokio::test]
-    async fn test_backend_info() {
-        let (backend, key_id, _key) = create_test_backend().await;
-
-        let info = KmsClient::backend_info(&backend);
-        assert_eq!(info.backend_type, "static");
-        assert_eq!(info.endpoint, "local");
-        assert!(info.healthy);
-        assert_eq!(info.metadata.get("key_id"), Some(&key_id));
+        KmsBackendTrait::health_check(&backend)
+            .await
+            .expect("health_check should succeed");
     }
 
     #[tokio::test]
@@ -740,17 +679,13 @@ mod tests {
 
         let plaintext = b"Hello, static KMS world!";
         let enc_request = EncryptRequest::new(key_id.clone(), plaintext.to_vec());
-        let enc_response = KmsClient::encrypt(&backend, &enc_request, None)
-            .await
-            .expect("encrypt should succeed");
+        let enc_response = backend.encrypt_to_envelope(&enc_request).expect("encrypt should succeed");
 
         assert_eq!(enc_response.key_id, key_id);
         assert!(!enc_response.ciphertext.is_empty());
 
         let dec_request = DecryptRequest::new(enc_response.ciphertext);
-        let decrypted = KmsClient::decrypt(&backend, &dec_request, None)
-            .await
-            .expect("decrypt should succeed");
+        let decrypted = backend.decrypt_envelope(&dec_request).expect("decrypt should succeed");
 
         assert_eq!(decrypted, plaintext);
     }

crates/kms/src/backends/vault.rs

@@ -19,8 +19,7 @@ use crate::backends::vault_credentials::{
     token_source_for,
 };
 use crate::backends::{
-    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits,
-    ensure_key_status_permits,
+    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits, ensure_key_status_permits,
 };
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -42,7 +41,6 @@ use vaultrs::{api::kv2::requests::SetSecretRequestOptions, error::ClientError, k
 /// Vault KMS client implementation
 pub struct VaultKmsClient {
     credentials: Arc<VaultCredentialProvider>,
-    config: VaultConfig,
     /// Mount path for the KV engine (typically "kv" or "secret")
     kv_mount: String,
     /// Path prefix for storing keys
@@ -200,7 +198,6 @@ impl VaultKmsClient {
             credentials,
             kv_mount: config.kv_mount.clone(),
             key_path_prefix: config.key_path_prefix.clone(),
-            config,
             dek_crypto: AesDekCrypto::new(),
             retry: RetryPolicy::from_config(kms_config),
             cancel: CancellationToken::new(),
@@ -254,13 +251,6 @@ impl VaultKmsClient {
         Ok(general_purpose::STANDARD.encode(key_material))
     }
 
-    /// Decode key material from KV2 storage (plain Base64, see `encrypt_key_material`).
-    async fn decrypt_key_material(&self, encrypted_material: &str) -> Result<Vec<u8>> {
-        general_purpose::STANDARD
-            .decode(encrypted_material)
-            .map_err(|e| KmsError::cryptographic_error("decrypt", e.to_string()))
-    }
-
     /// Read the immutable material record of one key version.
     ///
     /// A missing record fails closed with [`KmsError::KeyVersionNotFound`]; falling
@@ -589,9 +579,12 @@ impl VaultKmsClient {
     }
 }
 
-#[async_trait]
-impl KmsClient for VaultKmsClient {
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+impl VaultKmsClient {
+    pub(crate) async fn generate_data_key(
+        &self,
+        request: &GenerateKeyRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<DataKeyInfo> {
         debug!("Generating data key for master key: {}", request.master_key_id);
 
         let key_data = self.get_key_data(&request.master_key_id).await?;
@@ -632,20 +625,32 @@ impl KmsClient for VaultKmsClient {
         Ok(data_key)
     }
 
-    async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    pub(crate) async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
         debug!("Encrypting data with key: {}", request.key_id);
 
-        // Get the master key and verify its state allows encryption
+        // Single read of the key record: the material we wrap with and the
+        // version stamped into the envelope must come from the same snapshot
+        // (see generate_data_key).
         let key_data = self.get_key_data(&request.key_id).await?;
         ensure_key_status_permits(&request.key_id, &key_data.status, StateGatedOperation::Encrypt)?;
-        let key_material = self.decrypt_key_material(&key_data.encrypted_key_material).await?;
+        let key_material = decode_stored_key_material(&request.key_id, &key_data.encrypted_key_material)
+            .inspect_err(|error| warn!(key_id = %request.key_id, %error, "Vault KMS key material failed validation"))?;
+        let (encrypted_key, nonce) = self.dek_crypto.encrypt(&key_material, &request.plaintext).await?;
 
-        // For simplicity, we'll use a basic encryption approach
-        // In practice, you'd use proper AEAD encryption
-        let mut ciphertext = request.plaintext.clone();
-        for (i, byte) in ciphertext.iter_mut().enumerate() {
-            *byte ^= key_material[i % key_material.len()];
-        }
+        // Wrap the ciphertext in the same authenticated envelope that
+        // generate_data_key emits, so decrypt() round-trips it and resolves
+        // the wrapping master key version after rotations.
+        let envelope = DataKeyEnvelope {
+            key_id: uuid::Uuid::new_v4().to_string(),
+            master_key_id: request.key_id.clone(),
+            key_spec: "AES_256".to_string(),
+            encrypted_key,
+            nonce,
+            encryption_context: request.encryption_context.clone(),
+            created_at: Zoned::now(),
+            master_key_version: Some(key_data.version),
+        };
+        let ciphertext = serde_json::to_vec(&envelope)?;
 
         Ok(EncryptResponse {
             ciphertext,
@@ -655,7 +660,7 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    pub(crate) async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
         debug!("Decrypting data");
 
         // Parse the data key envelope from ciphertext
@@ -697,7 +702,12 @@ impl KmsClient for VaultKmsClient {
         Ok(plaintext)
     }
 
-    async fn create_key(&self, key_id: &str, algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    pub(crate) async fn create_key(
+        &self,
+        key_id: &str,
+        algorithm: &str,
+        _context: Option<&OperationContext>,
+    ) -> Result<MasterKeyInfo> {
         debug!("Creating master key: {} with algorithm: {}", key_id, algorithm);
 
         // Existence pre-check with read-confirm recovery: a create whose
@@ -779,7 +789,7 @@ impl KmsClient for VaultKmsClient {
         Ok(master_key)
     }
 
-    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
+    pub(crate) async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
         debug!("Describing key: {}", key_id);
 
         let key_data = self.get_key_data(key_id).await?;
@@ -799,7 +809,11 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
+    pub(crate) async fn list_keys(
+        &self,
+        request: &ListKeysRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<ListKeysResponse> {
         debug!("Listing keys with limit: {:?}", request.limit);
 
         let all_keys = self.list_vault_keys().await?;
@@ -836,7 +850,7 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Enabling key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
@@ -848,7 +862,7 @@ impl KmsClient for VaultKmsClient {
         Ok(())
     }
 
-    async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         debug!("Disabling key: {}", key_id);
 
         let mut key_data = self.get_key_data(key_id).await?;
@@ -860,39 +874,6 @@ impl KmsClient for VaultKmsClient {
         Ok(())
     }
 
-    async fn schedule_key_deletion(
-        &self,
-        key_id: &str,
-        pending_window_days: u32,
-        _context: Option<&OperationContext>,
-    ) -> Result<()> {
-        debug!("Scheduling key deletion: {}", key_id);
-
-        let mut key_data = self.get_key_data(key_id).await?;
-        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::ScheduleDeletion)?;
-        key_data.status = KeyStatus::PendingDeletion;
-        key_data.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400));
-        self.store_key_data(key_id, &key_data).await?;
-
-        debug!(key_id, "Vault KMS key deletion scheduled");
-        Ok(())
-    }
-
-    async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        debug!("Canceling key deletion: {}", key_id);
-
-        let mut key_data = self.get_key_data(key_id).await?;
-        if key_data.status != KeyStatus::PendingDeletion {
-            return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
-        }
-        key_data.status = KeyStatus::Active;
-        key_data.deletion_date = None;
-        self.store_key_data(key_id, &key_data).await?;
-
-        debug!(key_id, "Vault KMS key deletion canceled");
-        Ok(())
-    }
-
     /// Rotate the master key while keeping every historical version decryptable.
     ///
     /// Commit protocol (all writes check-and-set, in this order):
@@ -908,10 +889,11 @@ impl KmsClient for VaultKmsClient {
     /// or interrupted rotation never exposes half-committed material. Concurrent
     /// rotations are serialized by the check-and-set writes: at most one caller
     /// commits each version and the losers fail without side effects on current.
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         debug!("Rotating master key: {}", key_id);
 
         let (mut cas, mut key_data) = self.get_key_data_versioned(key_id).await?;
+        ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::Rotate)?;
 
         // The material about to be frozen must be decodable: freezing poisoned
         // material would give legacy envelopes a permanently broken baseline. This
@@ -992,7 +974,7 @@ impl KmsClient for VaultKmsClient {
         })
     }
 
-    async fn health_check(&self) -> Result<()> {
+    pub(crate) async fn health_check(&self) -> Result<()> {
         debug!("Performing Vault health check");
 
         // Use list_vault_keys but handle the case where no keys exist (which is normal)
@@ -1014,15 +996,6 @@ impl KmsClient for VaultKmsClient {
             }
         }
     }
-
-    fn backend_info(&self) -> BackendInfo {
-        BackendInfo::new("vault-kv2".to_string(), "0.1.0".to_string(), self.config.address.clone(), true)
-            .with_metadata("kv_mount".to_string(), self.kv_mount.clone())
-            .with_metadata("key_prefix".to_string(), self.key_path_prefix.clone())
-            // Master key material is protected only by Vault ACLs and KV2 at-rest
-            // encryption; there is no additional cryptographic wrapping.
-            .with_metadata("at_rest_protection".to_string(), "vault-kv2-acl".to_string())
-    }
 }
 
 /// VaultKmsBackend wraps VaultKmsClient and implements the KmsBackend trait
@@ -1031,12 +1004,6 @@ pub struct VaultKmsBackend {
 }
 
 impl VaultKmsBackend {
-    /// Lifecycle driver for the shared state-machine contract tests.
-    #[cfg(test)]
-    pub(crate) fn lifecycle_client(&self) -> &VaultKmsClient {
-        &self.client
-    }
-
     /// Create a new VaultKmsBackend
     pub async fn new(config: KmsConfig) -> Result<Self> {
         config.validate()?;
@@ -1296,17 +1263,32 @@ impl KmsBackend for VaultKmsBackend {
         })
     }
 
+    async fn enable_key(&self, key_id: &str) -> Result<()> {
+        self.client.enable_key(key_id, None).await
+    }
+
+    async fn disable_key(&self, key_id: &str) -> Result<()> {
+        self.client.disable_key(key_id, None).await
+    }
+
+    async fn rotate_key(&self, key_id: &str) -> Result<()> {
+        self.client.rotate_key(key_id, None).await.map(|_| ())
+    }
+
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
 
     fn capabilities(&self) -> BackendCapabilities {
-        // Rotation is unadvertised: the KV2 backend cannot rotate without
-        // replacing key material in place, and no historical versions are
-        // retained, so versioning is unsupported as well.
+        // Rotation freezes the outgoing material as an immutable version
+        // record before switching the current pointer, and envelopes resolve
+        // their wrapping version on decrypt, so every historical version
+        // stays decryptable after a rotation.
         BackendCapabilities::minimal()
+            .with_rotate(true)
             .with_enable_disable(true)
             .with_schedule_deletion(true)
+            .with_versioning(true)
             .with_physical_delete(true)
     }
 
@@ -1354,6 +1336,8 @@ mod tests {
     use crate::backends::scripted_vault::{ScriptedResponse, ScriptedVault};
     use crate::config::{VaultAuthMethod, VaultConfig};
 
+    const SCRIPTED_RETRY_ATTEMPTS: u32 = 3;
+
     /// Vault + KMS config pair pointing at a scripted loopback Vault.
     fn scripted_configs(address: &str) -> (VaultConfig, KmsConfig) {
         let vault_config = VaultConfig {
@@ -1369,7 +1353,7 @@ mod tests {
         };
         let kms_config = KmsConfig {
             timeout: Duration::from_secs(5),
-            retry_attempts: 3,
+            retry_attempts: SCRIPTED_RETRY_ATTEMPTS,
             ..KmsConfig::default()
         };
         (vault_config, kms_config)
@@ -1438,6 +1422,31 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn wired_read_retries_closed_connections_within_budget() {
+        let (vault, client) = scripted_client(vec![ScriptedResponse::close(), ScriptedResponse::close()]).await;
+
+        let error = client
+            .get_key_data("wired-key")
+            .await
+            .expect_err("closed connections must exhaust the retry budget");
+        assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        let expected_requests = usize::try_from(SCRIPTED_RETRY_ATTEMPTS).expect("retry attempts must fit usize");
+        assert_eq!(
+            requests.len(),
+            expected_requests,
+            "all budgeted retry attempts must reach Vault: {requests:?}"
+        );
+        assert!(
+            requests
+                .iter()
+                .all(|line| line == "GET /v1/secret/data/rustfs/kms/keys/wired-key"),
+            "all attempts must hit the same read endpoint: {requests:?}"
+        );
+    }
+
     #[tokio::test]
     async fn wired_read_does_not_retry_permission_errors() {
         let (vault, client) = scripted_client(vec![ScriptedResponse::error(403, "permission denied")]).await;
@@ -1720,19 +1729,6 @@ mod tests {
         assert!(!is_cas_conflict(&not_found));
     }
 
-    #[tokio::test]
-    async fn test_vault_kv2_backend_info_reports_at_rest_protection() {
-        let client = VaultKmsClient::new(integration_vault_config(), &KmsConfig::default())
-            .await
-            .expect("client");
-
-        let info = client.backend_info();
-        assert_eq!(info.backend_type, "vault-kv2");
-        assert_eq!(info.metadata.get("at_rest_protection").map(String::as_str), Some("vault-kv2-acl"));
-        // The KV2 backend must not present itself as Transit-backed.
-        assert!(!format!("{info:?}").contains("Transit"));
-    }
-
     fn integration_generate_request(key_id: &str) -> GenerateKeyRequest {
         GenerateKeyRequest {
             master_key_id: key_id.to_string(),
@@ -2081,4 +2077,150 @@ mod tests {
         let legacy: VaultKeyData = serde_json::from_value(value).expect("legacy record must deserialize");
         assert!(legacy.deletion_date.is_none());
     }
+
+    /// KV2 write acknowledgement (`SecretVersionMetadata`) for `kv2::set`.
+    fn kv2_write_ack() -> serde_json::Value {
+        serde_json::json!({
+            "created_time": "2026-01-01T00:00:00Z",
+            "custom_metadata": null,
+            "deletion_time": "",
+            "destroyed": false,
+            "version": 2,
+        })
+    }
+
+    #[tokio::test]
+    async fn wired_kv2_encrypt_round_trips_through_decrypt() {
+        // One key-record read for the encrypt, one for the decrypt.
+        let (_vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+        ])
+        .await;
+        let context = HashMap::from([("bucket".to_string(), "kv2".to_string())]);
+
+        let encrypted = client
+            .encrypt(
+                &EncryptRequest {
+                    key_id: "wired-key".to_string(),
+                    plaintext: b"kv2-direct-encrypt".to_vec(),
+                    encryption_context: context.clone(),
+                    grant_tokens: Vec::new(),
+                },
+                None,
+            )
+            .await
+            .expect("encrypt must produce an envelope");
+
+        // The ciphertext is a real KMS envelope wrapping AEAD output that
+        // decrypt() can open, not an XOR of the plaintext with the master key
+        // material.
+        let envelope: DataKeyEnvelope = serde_json::from_slice(&encrypted.ciphertext).expect("envelope must parse");
+        assert_eq!(envelope.master_key_id, "wired-key");
+        assert_eq!(envelope.master_key_version, Some(1));
+
+        let decrypted = client
+            .decrypt(
+                &DecryptRequest {
+                    ciphertext: encrypted.ciphertext.clone(),
+                    encryption_context: context,
+                    grant_tokens: Vec::new(),
+                },
+                None,
+            )
+            .await
+            .expect("decrypt must round-trip the envelope");
+        assert_eq!(decrypted, b"kv2-direct-encrypt".to_vec());
+
+        // A different object context must not decrypt (checked before any
+        // Vault read, so no scripted response is consumed).
+        let error = client
+            .decrypt(
+                &DecryptRequest {
+                    ciphertext: encrypted.ciphertext,
+                    encryption_context: HashMap::from([("bucket".to_string(), "other".to_string())]),
+                    grant_tokens: Vec::new(),
+                },
+                None,
+            )
+            .await
+            .expect_err("a different context must not decrypt");
+        assert!(matches!(error, KmsError::ContextMismatch { .. }), "got {error:?}");
+    }
+
+    /// KV2 secret-metadata read payload (`kv2::read_metadata`) pinning the
+    /// current secret version used as the rotation check-and-set base.
+    fn kv2_metadata_read_data(current_version: u64) -> serde_json::Value {
+        serde_json::json!({
+            "cas_required": false,
+            "created_time": "2026-01-01T00:00:00Z",
+            "current_version": current_version,
+            "delete_version_after": "0s",
+            "max_versions": 0,
+            "oldest_version": 0,
+            "updated_time": "2026-01-01T00:00:00Z",
+            "custom_metadata": null,
+            "versions": {},
+        })
+    }
+
+    #[tokio::test]
+    async fn wired_kv2_rotate_rejected_while_disabled() {
+        let mut key_data = healthy_key_data();
+        key_data.status = KeyStatus::Disabled;
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(kv2_metadata_read_data(1)),
+            ScriptedResponse::ok(kv2_read_data(&key_data)),
+        ])
+        .await;
+
+        let error = client
+            .rotate_key("wired-key", None)
+            .await
+            .expect_err("rotation of a disabled key must be rejected");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        assert_eq!(
+            requests.len(),
+            2,
+            "the state gate must reject after the versioned read, before any write: {requests:?}"
+        );
+        assert!(requests.iter().all(|line| line.starts_with("GET ")), "{requests:?}");
+    }
+
+    #[tokio::test]
+    async fn wired_backend_lifecycle_overrides_reach_the_client() {
+        let mut disabled = healthy_key_data();
+        disabled.status = KeyStatus::Disabled;
+        let vault = ScriptedVault::serve(vec![
+            // disable: read the Active record, persist it Disabled.
+            ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+            ScriptedResponse::ok(kv2_write_ack()),
+            // enable: read the Disabled record, persist it Active.
+            ScriptedResponse::ok(kv2_read_data(&disabled)),
+            ScriptedResponse::ok(kv2_write_ack()),
+        ])
+        .await;
+        let config = KmsConfig::vault(
+            url::Url::parse(&vault.address).expect("scripted vault address should parse"),
+            "scripted-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        let backend = VaultKmsBackend::new(config).await.expect("vault kv2 backend should build");
+
+        backend
+            .disable_key("wired-key")
+            .await
+            .expect("KmsBackend::disable_key must persist through the client");
+        backend
+            .enable_key("wired-key")
+            .await
+            .expect("KmsBackend::enable_key must persist through the client");
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 4, "each transition is one read plus one write: {requests:?}");
+        assert!(requests[0].starts_with("GET ") && requests[2].starts_with("GET "), "{requests:?}");
+        assert!(requests[1].starts_with("POST ") && requests[3].starts_with("POST "), "{requests:?}");
+    }
 }

crates/kms/src/backends/vault_transit.rs

@@ -18,9 +18,7 @@ use crate::backends::vault_credentials::{
     CredentialTaskHandle, VaultClientHandle, VaultConnectionSettings, VaultCredentialPolicy, VaultCredentialProvider,
     token_source_for,
 };
-use crate::backends::{
-    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits,
-};
+use crate::backends::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits};
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -506,9 +504,12 @@ impl VaultTransitKmsClient {
     }
 }
 
-#[async_trait]
-impl KmsClient for VaultTransitKmsClient {
-    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
+impl VaultTransitKmsClient {
+    pub(crate) async fn generate_data_key(
+        &self,
+        request: &GenerateKeyRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<DataKeyInfo> {
         self.ensure_key_state_allows(&request.master_key_id, StateGatedOperation::GenerateDataKey)
             .await?;
 
@@ -540,7 +541,7 @@ impl KmsClient for VaultTransitKmsClient {
         ))
     }
 
-    async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    pub(crate) async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
         let metadata = self
             .ensure_key_state_allows(&request.key_id, StateGatedOperation::Encrypt)
             .await?;
@@ -556,7 +557,7 @@ impl KmsClient for VaultTransitKmsClient {
         })
     }
 
-    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    pub(crate) async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
         let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
             .map_err(|e| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {e}")))?;
 
@@ -578,7 +579,14 @@ impl KmsClient for VaultTransitKmsClient {
             .await
     }
 
-    async fn create_key(&self, key_id: &str, algorithm: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn create_key(
+        &self,
+        key_id: &str,
+        algorithm: &str,
+        _context: Option<&OperationContext>,
+    ) -> Result<MasterKeyInfo> {
         if algorithm != "AES_256" {
             return Err(KmsError::unsupported_algorithm(algorithm));
         }
@@ -645,11 +653,17 @@ impl KmsClient for VaultTransitKmsClient {
         })
     }
 
-    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
         self.key_info(key_id).await
     }
 
-    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
+    pub(crate) async fn list_keys(
+        &self,
+        request: &ListKeysRequest,
+        _context: Option<&OperationContext>,
+    ) -> Result<ListKeysResponse> {
         let all_keys = self
             .run("vault_transit_list_keys", OpClass::ReadIdempotent, move || async move {
                 let vault = self.vault().map_err(AttemptError::fatal)?;
@@ -692,7 +706,7 @@ impl KmsClient for VaultTransitKmsClient {
         })
     }
 
-    async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         // A pending deletion must be reverted through cancel_key_deletion, not
         // silently by enabling, so the gate rejects PendingDeletion here.
         let mut metadata = self.ensure_key_state_allows(key_id, StateGatedOperation::Enable).await?;
@@ -701,13 +715,15 @@ impl KmsClient for VaultTransitKmsClient {
         self.store_key_metadata(key_id, &metadata).await
     }
 
-    async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
         let mut metadata = self.ensure_key_state_allows(key_id, StateGatedOperation::Disable).await?;
         metadata.key_state = KeyState::Disabled;
         self.store_key_metadata(key_id, &metadata).await
     }
 
-    async fn schedule_key_deletion(
+    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
+    #[cfg(test)]
+    pub(crate) async fn schedule_key_deletion(
         &self,
         key_id: &str,
         pending_window_days: u32,
@@ -721,17 +737,7 @@ impl KmsClient for VaultTransitKmsClient {
         self.store_key_metadata(key_id, &metadata).await
     }
 
-    async fn cancel_key_deletion(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
-        let mut metadata = self.get_key_metadata(key_id).await?;
-        if metadata.key_state != KeyState::PendingDeletion {
-            return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion")));
-        }
-        metadata.key_state = KeyState::Enabled;
-        metadata.deletion_date = None;
-        self.store_key_metadata(key_id, &metadata).await
-    }
-
-    async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
         self.ensure_key_state_allows(key_id, StateGatedOperation::Rotate).await?;
 
         // Single attempt, never retried: replaying a rotate whose response was
@@ -768,7 +774,7 @@ impl KmsClient for VaultTransitKmsClient {
         })
     }
 
-    async fn health_check(&self) -> Result<()> {
+    pub(crate) async fn health_check(&self) -> Result<()> {
         self.run("vault_transit_health_check", OpClass::ReadIdempotent, move || async move {
             let vault = self.vault().map_err(AttemptError::fatal)?;
             key::list(&vault.client, &self.config.mount_path)
@@ -780,11 +786,6 @@ impl KmsClient for VaultTransitKmsClient {
         })
         .await
     }
-
-    fn backend_info(&self) -> BackendInfo {
-        BackendInfo::new("vault-transit".to_string(), "0.1.0".to_string(), self.config.address.clone(), true)
-            .with_metadata("mount_path".to_string(), self.config.mount_path.clone())
-    }
 }
 
 pub struct VaultTransitKmsBackend {
@@ -792,14 +793,6 @@ pub struct VaultTransitKmsBackend {
 }
 
 impl VaultTransitKmsBackend {
-    /// Lifecycle driver for the shared state-machine contract tests. Using the
-    /// backend's own client keeps its in-process metadata cache coherent with
-    /// the transitions the tests perform.
-    #[cfg(test)]
-    pub(crate) fn lifecycle_client(&self) -> &VaultTransitKmsClient {
-        &self.client
-    }
-
     pub async fn new(config: KmsConfig) -> Result<Self> {
         config.validate()?;
 
@@ -1000,6 +993,18 @@ impl KmsBackend for VaultTransitKmsBackend {
         })
     }
 
+    async fn enable_key(&self, key_id: &str) -> Result<()> {
+        self.client.enable_key(key_id, None).await
+    }
+
+    async fn disable_key(&self, key_id: &str) -> Result<()> {
+        self.client.disable_key(key_id, None).await
+    }
+
+    async fn rotate_key(&self, key_id: &str) -> Result<()> {
+        self.client.rotate_key(key_id, None).await.map(|_| ())
+    }
+
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
@@ -1437,4 +1442,58 @@ mod tests {
         assert_eq!(metadata.key_state, KeyState::Enabled);
         assert!(metadata.deletion_date.is_none());
     }
+
+    /// KV2 write acknowledgement (`SecretVersionMetadata`) for `kv2::set`.
+    fn kv2_write_ack() -> serde_json::Value {
+        serde_json::json!({
+            "created_time": "2026-01-01T00:00:00Z",
+            "custom_metadata": null,
+            "deletion_time": "",
+            "destroyed": false,
+            "version": 2,
+        })
+    }
+
+    #[tokio::test]
+    async fn wired_backend_lifecycle_overrides_reach_the_client() {
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        let vault = ScriptedVault::serve(vec![
+            // disable: metadata cache miss reads KV, then persists Disabled.
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ScriptedResponse::ok(kv2_write_ack()),
+            // enable: the state gate hits the metadata cache, so only the
+            // persisting write goes out.
+            ScriptedResponse::ok(kv2_write_ack()),
+            // rotate: the gate hits the cache again; the single rotate
+            // attempt fails and must not be retried.
+            ScriptedResponse::error(503, "standby"),
+        ])
+        .await;
+        let config = KmsConfig::vault_transit(
+            url::Url::parse(&vault.address).expect("scripted vault address should parse"),
+            "scripted-token".to_string(),
+        )
+        .with_insecure_development_defaults();
+        let backend = VaultTransitKmsBackend::new(config)
+            .await
+            .expect("vault transit backend should build");
+
+        backend
+            .disable_key("wired-key")
+            .await
+            .expect("KmsBackend::disable_key must persist through the client");
+        backend
+            .enable_key("wired-key")
+            .await
+            .expect("KmsBackend::enable_key must persist through the client");
+        let error = backend
+            .rotate_key("wired-key")
+            .await
+            .expect_err("the scripted 503 must fail the rotation");
+        assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        assert_eq!(requests.len(), 4, "gated reads, two writes and one rotate attempt: {requests:?}");
+        assert_eq!(requests[3], "POST /v1/transit/keys/wired-key/rotate", "{requests:?}");
+    }
 }

crates/kms/src/backup/local_export.rs

@@ -0,0 +1,998 @@
+// 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.
+
+//! Local backend backup export: sealed, KEK-protected bundle production.
+//!
+//! This is the producer side only; restore lives in a follow-up change. The
+//! admin API is not wired here either — callers construct the request and
+//! supply the backup KEK explicitly.
+//!
+//! # Bundle layout
+//!
+//! A bundle is a directory (simple to produce, artifacts stream one file at a
+//! time, and partial output is trivially recognizable because the manifest is
+//! written last):
+//!
+//! ```text
+//! <destination>/
+//!   manifest.json                      # sealed BackupManifest, written last
+//!   artifacts/keys/<key_id>.key.enc    # one per stored key record
+//!   artifacts/master-key.salt.enc      # present when the salt file exists
+//! ```
+//!
+//! # Artifact payload framing
+//!
+//! Every artifact payload is `nonce (12 bytes) || AES-256-GCM ciphertext`,
+//! encrypted under the caller-supplied backup KEK with an AAD binding of
+//! `(context, backup_id, snapshot_generation, artifact path)`, so an artifact
+//! cannot be swapped into another bundle or renamed within its own bundle.
+//! Records already encrypted at rest stay encrypted inside the wrap;
+//! plaintext-dev-only records become ciphertext-only in the bundle, which is
+//! their mandatory re-wrap under the backup KEK.
+//!
+//! # Write protocol
+//!
+//! Artifacts are written and fsynced first, re-read and digest-verified, and
+//! only then is the sealed manifest (completeness marker plus final digest)
+//! published. A crash at any earlier point leaves a bundle without a
+//! manifest, which decodes as an incomplete bundle and can never be restored.
+
+use crate::backends::local::{LocalKmsClient, StoredKeyProtection};
+use crate::backup::capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
+use crate::backup::error::BackupError;
+use crate::backup::manifest::{
+    AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest,
+    DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation,
+};
+use crate::error::{KmsError, Result};
+use aes_gcm::{
+    Aes256Gcm, Key, Nonce,
+    aead::{Aead, KeyInit, Payload},
+};
+use jiff::Zoned;
+use rand::RngExt;
+use serde::Deserialize;
+use std::path::{Path, PathBuf};
+use tokio::fs;
+use tokio::io::AsyncWriteExt;
+use zeroize::Zeroizing;
+
+/// File name of the sealed manifest inside a bundle directory.
+pub const LOCAL_BUNDLE_MANIFEST_FILE: &str = "manifest.json";
+const ARTIFACTS_DIR: &str = "artifacts";
+const KEYS_DIR: &str = "artifacts/keys";
+const SALT_ARTIFACT_PATH: &str = "artifacts/master-key.salt.enc";
+const AEAD_NONCE_LEN: usize = 12;
+/// Domain-separation context for the artifact AAD binding.
+const BUNDLE_AAD_CONTEXT: &str = "rustfs-kms-local-backup:v1";
+
+/// Caller-supplied backup KEK: a trust root deliberately separate from the
+/// business KMS hierarchy (it must not be a key that is itself part of the
+/// state being backed up). Where the KEK comes from is the admin layer's
+/// concern; this module only consumes it.
+pub struct BackupKek {
+    kek_id: String,
+    kek_version: u32,
+    key: Zeroizing<[u8; 32]>,
+}
+
+impl BackupKek {
+    /// Wrap 32 bytes of KEK material. The material is zeroized on drop;
+    /// callers should zeroize their own copy of the input.
+    pub fn new(kek_id: impl Into<String>, kek_version: u32, key: [u8; 32]) -> Result<Self> {
+        let kek_id = kek_id.into();
+        if kek_id.is_empty() {
+            return Err(KmsError::validation_error("backup KEK id must not be empty"));
+        }
+        Ok(Self {
+            kek_id,
+            kek_version,
+            key: Zeroizing::new(key),
+        })
+    }
+
+    /// Manifest descriptor for this KEK.
+    pub fn descriptor(&self) -> BackupKekDescriptor {
+        BackupKekDescriptor {
+            kek_id: self.kek_id.clone(),
+            kek_version: self.kek_version,
+            aead_algorithm: AeadAlgorithm::Aes256Gcm,
+        }
+    }
+
+    fn cipher(&self) -> Aes256Gcm {
+        Aes256Gcm::new(&Key::<Aes256Gcm>::from(*self.key))
+    }
+}
+
+/// Parameters of one export run.
+///
+/// `snapshot_generation` is injected by the caller: the contract only
+/// requires it to be monotonic per deployment, and the source (persisted
+/// counter, coordinated clock) is decided by the admin layer, which keeps
+/// this module free of ambient time or state lookups.
+#[derive(Debug, Clone)]
+pub struct LocalBackupExportRequest {
+    /// Unique identifier for this backup.
+    pub backup_id: String,
+    /// Opaque identity of the producing deployment.
+    pub deployment_identity: String,
+    /// RustFS version string recorded in the manifest.
+    pub rustfs_version: String,
+    /// Monotonic snapshot generation this bundle belongs to.
+    pub snapshot_generation: u64,
+    /// Bundle output directory; must not exist yet or must be empty.
+    pub destination: PathBuf,
+}
+
+impl LocalBackupExportRequest {
+    fn validate(&self) -> Result<()> {
+        for (field, value) in [
+            ("backup_id", &self.backup_id),
+            ("deployment_identity", &self.deployment_identity),
+            ("rustfs_version", &self.rustfs_version),
+        ] {
+            if value.is_empty() {
+                return Err(KmsError::validation_error(format!("backup export {field} must not be empty")));
+            }
+        }
+        Ok(())
+    }
+}
+
+/// Minimal projection of a stored key record: only the fields the exporter
+/// needs. Unknown fields are ignored on purpose — the record travels into the
+/// bundle byte-identical, so the exporter must not constrain its schema.
+#[derive(Deserialize)]
+struct StoredRecordProbe {
+    key_id: String,
+    #[serde(default)]
+    at_rest_protection: StoredKeyProtection,
+}
+
+struct CollectedRecord {
+    key_id: String,
+    protection: StoredKeyProtection,
+    /// Raw record bytes exactly as stored. Zeroized on drop because
+    /// plaintext-dev-only records embed key material.
+    raw: Zeroizing<Vec<u8>>,
+}
+
+struct CollectedSnapshot {
+    records: Vec<CollectedRecord>,
+    salt: Option<Vec<u8>>,
+}
+
+/// Export the Local backend's key directory as a sealed backup bundle.
+///
+/// The directory scan runs under the export fence, so concurrent
+/// create/update/delete operations are either fully included or fully
+/// excluded — never half a record. Encryption and bundle writing happen after
+/// the fence is released to keep it short.
+///
+/// Returns the sealed manifest that was written to the bundle.
+pub async fn export_local_backup(
+    client: &LocalKmsClient,
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+) -> Result<BackupManifest> {
+    request.validate()?;
+    prepare_destination(&request.destination).await?;
+
+    let snapshot = collect_snapshot(client).await?;
+    if snapshot.records.is_empty() {
+        return Err(KmsError::invalid_operation(
+            "Local backup export found no key records; refusing to publish an empty bundle",
+        ));
+    }
+
+    let has_encrypted = snapshot
+        .records
+        .iter()
+        .any(|record| record.protection == StoredKeyProtection::EncryptedMasterKey);
+    if has_encrypted && snapshot.salt.is_none() {
+        return Err(KmsError::invalid_operation(
+            "key directory contains encrypted-master-key records but the master key salt file is missing; \
+             the bundle would be unrestorable",
+        ));
+    }
+
+    let manifest = build_and_write_bundle(kek, request, &snapshot).await?;
+    Ok(manifest)
+}
+
+/// Read and fully validate the manifest of a local bundle directory.
+///
+/// A directory without a manifest is an interrupted export: the manifest is
+/// written last, so its absence means the bundle never sealed.
+pub async fn read_local_bundle_manifest(bundle_dir: &Path) -> Result<BackupManifest> {
+    let manifest_path = bundle_dir.join(LOCAL_BUNDLE_MANIFEST_FILE);
+    let bytes = match fs::read(&manifest_path).await {
+        Ok(bytes) => bytes,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Err(BackupError::incomplete_bundle("bundle has no manifest; the export never sealed it").into());
+        }
+        Err(error) => return Err(error.into()),
+    };
+    let manifest = BackupManifest::decode(&bytes)?;
+    if manifest.backend != BackupBackendKind::Local {
+        return Err(
+            BackupError::corrupted(format!("bundle manifest declares backend {:?}, expected Local", manifest.backend)).into(),
+        );
+    }
+    Ok(manifest)
+}
+
+/// Read, verify, and decrypt one artifact of a local bundle.
+///
+/// Fail-closed order: KEK identity, artifact presence, declared length,
+/// encrypted digest, then AEAD authentication. The returned plaintext is
+/// zeroized on drop.
+pub async fn decrypt_bundle_artifact(
+    bundle_dir: &Path,
+    manifest: &BackupManifest,
+    descriptor: &ArtifactDescriptor,
+    kek: &BackupKek,
+) -> Result<Zeroizing<Vec<u8>>> {
+    manifest.backup_kek.ensure_matches(&kek.kek_id, kek.kek_version)?;
+    if descriptor.aead_algorithm != AeadAlgorithm::Aes256Gcm {
+        return Err(KmsError::unsupported_algorithm(format!(
+            "{:?} (local bundles are produced with AES-256-GCM)",
+            descriptor.aead_algorithm
+        )));
+    }
+
+    let artifact_path = bundle_dir.join(&descriptor.path);
+    let payload = match fs::read(&artifact_path).await {
+        Ok(payload) => payload,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Err(BackupError::missing_artifact(descriptor.path.clone()).into());
+        }
+        Err(error) => return Err(error.into()),
+    };
+
+    if (payload.len() as u64) < descriptor.len {
+        return Err(BackupError::truncated(format!(
+            "artifact '{}' is {} bytes, manifest declares {}",
+            descriptor.path,
+            payload.len(),
+            descriptor.len
+        ))
+        .into());
+    }
+    if payload.len() as u64 != descriptor.len {
+        return Err(BackupError::corrupted(format!(
+            "artifact '{}' is {} bytes, manifest declares {}",
+            descriptor.path,
+            payload.len(),
+            descriptor.len
+        ))
+        .into());
+    }
+    if ContentDigest::sha256_of(&payload) != descriptor.encrypted_digest {
+        return Err(BackupError::corrupted(format!("artifact '{}' does not match its manifest digest", descriptor.path)).into());
+    }
+    if payload.len() < AEAD_NONCE_LEN {
+        return Err(BackupError::corrupted(format!("artifact '{}' is too short to carry a nonce", descriptor.path)).into());
+    }
+
+    let (nonce_bytes, ciphertext) = payload.split_at(AEAD_NONCE_LEN);
+    let mut nonce = [0u8; AEAD_NONCE_LEN];
+    nonce.copy_from_slice(nonce_bytes);
+    let aad = artifact_aad(&manifest.backup_id, manifest.snapshot_generation, &descriptor.path);
+    let plaintext = kek
+        .cipher()
+        .decrypt(
+            &Nonce::from(nonce),
+            Payload {
+                msg: ciphertext,
+                aad: &aad,
+            },
+        )
+        .map_err(|_| {
+            KmsError::from(BackupError::corrupted(format!(
+                "artifact '{}' failed authenticated decryption under the supplied backup KEK",
+                descriptor.path
+            )))
+        })?;
+    Ok(Zeroizing::new(plaintext))
+}
+
+/// Scan the key directory under the export fence.
+async fn collect_snapshot(client: &LocalKmsClient) -> Result<CollectedSnapshot> {
+    let _fence = client.acquire_export_fence().await;
+
+    let mut records = Vec::new();
+    let mut entries = fs::read_dir(client.key_directory()).await?;
+    while let Some(entry) = entries.next_entry().await? {
+        let path = entry.path();
+        if !path.extension().is_some_and(|extension| extension == "key") {
+            continue;
+        }
+        let stem = path
+            .file_stem()
+            .and_then(|stem| stem.to_str())
+            .ok_or_else(|| KmsError::configuration_error("Local KMS key file name must be valid UTF-8"))?
+            .to_string();
+
+        let raw = Zeroizing::new(fs::read(&path).await?);
+        // Any unreadable record aborts the export: a bundle silently missing
+        // one key is worse than no bundle at all.
+        let probe: StoredRecordProbe = serde_json::from_slice(&raw)
+            .map_err(|error| KmsError::material_corrupt(&stem, format!("stored key record does not deserialize: {error}")))?;
+        if probe.key_id != stem {
+            return Err(KmsError::invalid_key(format!(
+                "Local KMS key file identity mismatch: expected {stem:?}, found {:?}",
+                probe.key_id
+            )));
+        }
+
+        records.push(CollectedRecord {
+            key_id: stem,
+            protection: probe.at_rest_protection,
+            raw,
+        });
+    }
+
+    let salt_path = client.master_key_salt_file();
+    let salt = match fs::read(&salt_path).await {
+        Ok(bytes) => Some(bytes),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+        Err(error) => return Err(error.into()),
+    };
+
+    records.sort_by(|a, b| a.key_id.cmp(&b.key_id));
+    Ok(CollectedSnapshot { records, salt })
+}
+
+async fn build_and_write_bundle(
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+    snapshot: &CollectedSnapshot,
+) -> Result<BackupManifest> {
+    let mut artifacts = Vec::with_capacity(snapshot.records.len() + 1);
+    for record in &snapshot.records {
+        let artifact_path = format!("{KEYS_DIR}/{}.key.enc", record.key_id);
+        let descriptor = encrypt_and_write_artifact(kek, request, ArtifactKind::KeyMaterial, &artifact_path, &record.raw).await?;
+        artifacts.push(descriptor);
+    }
+    if let Some(salt) = &snapshot.salt {
+        let descriptor = encrypt_and_write_artifact(kek, request, ArtifactKind::MasterKeySalt, SALT_ARTIFACT_PATH, salt).await?;
+        artifacts.push(descriptor);
+    }
+
+    // Make the artifact directory entries durable before sealing: the sealed
+    // manifest must never survive a crash that its artifacts did not.
+    fsync_dir(&request.destination.join(KEYS_DIR)).await?;
+    fsync_dir(&request.destination.join(ARTIFACTS_DIR)).await?;
+
+    let manifest = BackupManifest {
+        format_version: BackupManifest::FORMAT_VERSION,
+        backup_id: request.backup_id.clone(),
+        // Normalized to UTC so the stored spelling is host-independent: the
+        // local zone's name (or a POSIX TZ string in minimal containers) has
+        // no business inside a portable bundle.
+        created_at: Zoned::now().with_time_zone(jiff::tz::TimeZone::UTC),
+        rustfs_version: request.rustfs_version.clone(),
+        deployment_identity: request.deployment_identity.clone(),
+        backend: BackupBackendKind::Local,
+        at_rest_protection: weakest_observed_protection(&snapshot.records),
+        responsibility: BackupResponsibility::FullMaterial,
+        snapshot_generation: request.snapshot_generation,
+        backup_kek: kek.descriptor(),
+        artifacts,
+        local_kdf: Some(local_kdf_descriptor(snapshot)),
+        key_versions: None,
+        capability_discovery: None,
+        completeness: CompletenessState::InProgress,
+        manifest_digest: ContentDigest {
+            algorithm: DigestAlgorithm::Sha256,
+            hex: String::new(),
+        },
+    };
+    let manifest = manifest.seal()?;
+    let manifest_bytes = manifest.encode()?;
+
+    write_new_file(&request.destination.join(LOCAL_BUNDLE_MANIFEST_FILE), &manifest_bytes).await?;
+    fsync_dir(&request.destination).await?;
+    Ok(manifest)
+}
+
+/// Encrypt one artifact, write it durably, and re-read it to verify the
+/// digest before it is allowed into the manifest.
+async fn encrypt_and_write_artifact(
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+    kind: ArtifactKind,
+    artifact_path: &str,
+    plaintext: &[u8],
+) -> Result<ArtifactDescriptor> {
+    let mut nonce = [0u8; AEAD_NONCE_LEN];
+    rand::rng().fill(&mut nonce[..]);
+    let aad = artifact_aad(&request.backup_id, request.snapshot_generation, artifact_path);
+    let ciphertext = kek
+        .cipher()
+        .encrypt(
+            &Nonce::from(nonce),
+            Payload {
+                msg: plaintext,
+                aad: &aad,
+            },
+        )
+        .map_err(|error| KmsError::cryptographic_error("backup_artifact_encrypt", error.to_string()))?;
+
+    let mut payload = Vec::with_capacity(AEAD_NONCE_LEN + ciphertext.len());
+    payload.extend_from_slice(&nonce);
+    payload.extend_from_slice(&ciphertext);
+
+    let absolute_path = request.destination.join(artifact_path);
+    write_new_file(&absolute_path, &payload).await?;
+
+    // Verify what actually landed on disk, not the in-memory buffer.
+    let written = fs::read(&absolute_path).await?;
+    let digest = ContentDigest::sha256_of(&written);
+    if written != payload {
+        return Err(KmsError::internal_error(format!(
+            "bundle artifact '{artifact_path}' read back differently than written"
+        )));
+    }
+
+    Ok(ArtifactDescriptor {
+        kind,
+        path: artifact_path.to_string(),
+        len: payload.len() as u64,
+        aead_algorithm: AeadAlgorithm::Aes256Gcm,
+        encrypted_digest: digest,
+    })
+}
+
+/// AAD binding an artifact to its bundle identity and path. A JSON tuple
+/// gives unambiguous field boundaries without a hand-rolled framing format.
+fn artifact_aad(backup_id: &str, snapshot_generation: u64, artifact_path: &str) -> Vec<u8> {
+    serde_json::to_vec(&(BUNDLE_AAD_CONTEXT, backup_id, snapshot_generation, artifact_path))
+        .expect("AAD tuple of strings and integers always serializes")
+}
+
+/// The bundle-level protection label is the weakest state observed across
+/// records: any plaintext-dev-only record marks the whole bundle, then any
+/// legacy-unspecified marker (unknown until read), and only a uniformly
+/// encrypted directory is labeled encrypted-master-key.
+fn weakest_observed_protection(records: &[CollectedRecord]) -> AtRestProtection {
+    let mut has_legacy = false;
+    for record in records {
+        match record.protection {
+            StoredKeyProtection::PlaintextDevOnly => return AtRestProtection::PlaintextDevOnly,
+            StoredKeyProtection::LegacyUnspecified => has_legacy = true,
+            StoredKeyProtection::EncryptedMasterKey => {}
+        }
+    }
+    if has_legacy {
+        AtRestProtection::LegacyUnspecified
+    } else {
+        AtRestProtection::EncryptedMasterKey
+    }
+}
+
+fn local_kdf_descriptor(snapshot: &CollectedSnapshot) -> LocalKdfDescriptor {
+    let mut modes = Vec::new();
+    for (marker, mode) in [
+        (StoredKeyProtection::EncryptedMasterKey, AtRestProtection::EncryptedMasterKey),
+        (StoredKeyProtection::PlaintextDevOnly, AtRestProtection::PlaintextDevOnly),
+        (StoredKeyProtection::LegacyUnspecified, AtRestProtection::LegacyUnspecified),
+    ] {
+        if snapshot.records.iter().any(|record| record.protection == marker) {
+            modes.push(mode);
+        }
+    }
+
+    // With a salt on disk the backend derives via Argon2id; without one only
+    // the pre-beta.9 SHA-256 derivation can apply. For plaintext-only
+    // directories the derivation is informational.
+    let derivation = if snapshot.salt.is_some() {
+        LocalKeyDerivation::current_argon2id()
+    } else {
+        LocalKeyDerivation::LegacySha256
+    };
+
+    LocalKdfDescriptor {
+        derivation,
+        protection_modes: modes,
+        // The verifier shape is left to the restore change; the schema keeps
+        // it optional so bundles without one stay valid.
+        master_key_verifier: None,
+    }
+}
+
+async fn prepare_destination(destination: &Path) -> Result<()> {
+    if fs::try_exists(destination).await? {
+        let mut entries = fs::read_dir(destination)
+            .await
+            .map_err(|error| KmsError::invalid_operation(format!("backup destination is not a readable directory: {error}")))?;
+        if entries.next_entry().await?.is_some() {
+            return Err(KmsError::invalid_operation(
+                "backup destination directory is not empty; refusing to mix bundles",
+            ));
+        }
+    }
+    fs::create_dir_all(destination.join(KEYS_DIR)).await?;
+    Ok(())
+}
+
+async fn write_new_file(path: &Path, bytes: &[u8]) -> Result<()> {
+    let mut file = fs::OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .open(path)
+        .await
+        .map_err(|error| KmsError::io_error(format!("failed to create bundle file {}: {error}", path.display())))?;
+    file.write_all(bytes).await?;
+    file.sync_all().await?;
+    Ok(())
+}
+
+/// Fsync a directory so freshly created bundle entries survive power loss.
+/// No-op on non-Unix platforms where directories cannot be opened for
+/// syncing (mirrors the local backend's durable commit helper).
+async fn fsync_dir(path: &Path) -> Result<()> {
+    #[cfg(unix)]
+    {
+        let path = path.to_path_buf();
+        tokio::task::spawn_blocking(move || std::fs::File::open(&path)?.sync_all())
+            .await
+            .map_err(|error| KmsError::io_error(error.to_string()))??;
+    }
+    #[cfg(not(unix))]
+    let _ = path;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::LocalConfig;
+    use std::sync::Arc;
+    use tempfile::TempDir;
+
+    async fn encrypted_client() -> (LocalKmsClient, TempDir) {
+        let temp = TempDir::new().expect("temp dir");
+        let client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp.path().to_path_buf(),
+            master_key: Some("test-master-key".to_string()),
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("client should initialize");
+        (client, temp)
+    }
+
+    async fn dev_client() -> (LocalKmsClient, TempDir) {
+        let temp = TempDir::new().expect("temp dir");
+        let client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp.path().to_path_buf(),
+            master_key: None,
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("client should initialize");
+        (client, temp)
+    }
+
+    fn test_kek() -> BackupKek {
+        BackupKek::new("backup-kek-test", 1, [0x42; 32]).expect("kek")
+    }
+
+    fn export_request(destination: PathBuf) -> LocalBackupExportRequest {
+        LocalBackupExportRequest {
+            backup_id: "backup-0001".to_string(),
+            deployment_identity: "deployment-test".to_string(),
+            rustfs_version: "1.0.0-test".to_string(),
+            snapshot_generation: 7,
+            destination,
+        }
+    }
+
+    fn walk_files(dir: &Path, out: &mut Vec<PathBuf>) {
+        for entry in std::fs::read_dir(dir).expect("read dir") {
+            let path = entry.expect("dir entry").path();
+            if path.is_dir() {
+                walk_files(&path, out);
+            } else {
+                out.push(path);
+            }
+        }
+    }
+
+    fn contains_subslice(haystack: &[u8], needle: &[u8]) -> bool {
+        !needle.is_empty() && haystack.windows(needle.len()).any(|window| window == needle)
+    }
+
+    #[tokio::test]
+    async fn export_round_trips_and_decrypts_to_source_records() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("alpha", "AES_256", None).await.expect("create alpha");
+        client.create_key("beta", "AES_256", None).await.expect("create beta");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.backend, BackupBackendKind::Local);
+        assert_eq!(manifest.responsibility, BackupResponsibility::FullMaterial);
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::EncryptedMasterKey);
+        assert_eq!(manifest.snapshot_generation, 7);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(kdf.derivation, LocalKeyDerivation::current_argon2id());
+        assert_eq!(kdf.protection_modes, vec![AtRestProtection::EncryptedMasterKey]);
+
+        // alpha, beta (sorted), then the salt artifact.
+        assert_eq!(manifest.artifacts.len(), 3);
+        assert_eq!(manifest.artifacts[0].path, "artifacts/keys/alpha.key.enc");
+        assert_eq!(manifest.artifacts[1].path, "artifacts/keys/beta.key.enc");
+        assert_eq!(manifest.artifacts[2].kind, ArtifactKind::MasterKeySalt);
+
+        let reread = read_local_bundle_manifest(&destination)
+            .await
+            .expect("manifest should decode");
+        assert_eq!(reread, manifest);
+
+        for (artifact, key_id) in [(&manifest.artifacts[0], "alpha"), (&manifest.artifacts[1], "beta")] {
+            let decrypted = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+                .await
+                .expect("artifact should decrypt");
+            let source = fs::read(client.key_directory().join(format!("{key_id}.key")))
+                .await
+                .expect("source record");
+            assert_eq!(decrypted.as_slice(), source.as_slice(), "record {key_id} must round-trip verbatim");
+        }
+
+        let salt = decrypt_bundle_artifact(&destination, &manifest, &manifest.artifacts[2], &kek)
+            .await
+            .expect("salt should decrypt");
+        let source_salt = fs::read(client.master_key_salt_file()).await.expect("source salt");
+        assert_eq!(salt.as_slice(), source_salt.as_slice());
+    }
+
+    #[tokio::test]
+    async fn plaintext_dev_only_material_is_rewrapped_and_absent_from_bundle() {
+        let (client, _key_dir) = dev_client().await;
+        client.create_key("dev-key", "AES_256", None).await.expect("create key");
+
+        let material = client
+            .decrypt_key_material_for_export("dev-key")
+            .await
+            .expect("material should be readable");
+        let source_record = fs::read(client.key_directory().join("dev-key.key")).await.expect("record");
+        let record_json: serde_json::Value = serde_json::from_slice(&source_record).expect("record parses");
+        let material_base64 = record_json
+            .get("encrypted_key_material")
+            .and_then(|value| value.as_str())
+            .expect("material field")
+            .to_string();
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::PlaintextDevOnly);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(kdf.protection_modes, vec![AtRestProtection::PlaintextDevOnly]);
+        assert_eq!(kdf.derivation, LocalKeyDerivation::LegacySha256);
+        assert!(
+            !manifest.artifacts.iter().any(|a| a.kind == ArtifactKind::MasterKeySalt),
+            "dev-mode directory has no salt to bundle"
+        );
+
+        // Byte-level: neither the raw material nor its base64 form may appear
+        // anywhere in the bundle. The mandatory KEK re-wrap is what hides it.
+        let mut files = Vec::new();
+        walk_files(&destination, &mut files);
+        assert!(!files.is_empty());
+        for file in files {
+            let bytes = std::fs::read(&file).expect("bundle file");
+            assert!(
+                !contains_subslice(&bytes, material.as_ref()),
+                "raw key material leaked into {}",
+                file.display()
+            );
+            assert!(
+                !contains_subslice(&bytes, material_base64.as_bytes()),
+                "base64 key material leaked into {}",
+                file.display()
+            );
+        }
+
+        // The wrapped record still round-trips for restore.
+        let decrypted = decrypt_bundle_artifact(&destination, &manifest, &manifest.artifacts[0], &kek)
+            .await
+            .expect("artifact should decrypt");
+        assert_eq!(decrypted.as_slice(), source_record.as_slice());
+    }
+
+    #[tokio::test]
+    async fn export_fence_blocks_writers_until_released() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("existing", "AES_256", None).await.expect("create key");
+        let client = Arc::new(client);
+
+        let fence = client.acquire_export_fence().await;
+
+        let writer = {
+            let client = Arc::clone(&client);
+            tokio::spawn(async move {
+                client.create_key("new-key", "AES_256", None).await.expect("create");
+                client.disable_key("existing", None).await.expect("disable");
+            })
+        };
+
+        for _ in 0..64 {
+            tokio::task::yield_now().await;
+        }
+        assert!(!writer.is_finished(), "writers must stay blocked while the export fence is held");
+
+        drop(fence);
+        writer.await.expect("writer should finish after fence release");
+        assert!(
+            fs::try_exists(client.key_directory().join("new-key.key"))
+                .await
+                .expect("exists")
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn concurrent_writers_yield_complete_records() {
+        let (client, _key_dir) = encrypted_client().await;
+        for index in 0..5 {
+            client
+                .create_key(&format!("seed-{index}"), "AES_256", None)
+                .await
+                .expect("seed key");
+        }
+        let client = Arc::new(client);
+
+        let writer = {
+            let client = Arc::clone(&client);
+            tokio::spawn(async move {
+                for index in 0..30 {
+                    client
+                        .create_key(&format!("concurrent-{index}"), "AES_256", None)
+                        .await
+                        .expect("create");
+                    let target = format!("seed-{}", index % 5);
+                    if index % 2 == 0 {
+                        client.disable_key(&target, None).await.expect("disable");
+                    } else {
+                        client.enable_key(&target, None).await.expect("enable");
+                    }
+                }
+            })
+        };
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed under concurrent writers");
+        writer.await.expect("writer task");
+
+        // Whatever subset of writers landed before the fence, every record in
+        // the bundle must be complete: parseable, self-identifying, and with
+        // non-empty material. No torn records, no half-updates.
+        let reread = read_local_bundle_manifest(&destination).await.expect("manifest decodes");
+        assert_eq!(reread, manifest);
+        for artifact in manifest.artifacts.iter().filter(|a| a.kind == ArtifactKind::KeyMaterial) {
+            let record = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+                .await
+                .expect("record decrypts");
+            let value: serde_json::Value = serde_json::from_slice(&record).expect("record is complete JSON");
+            let key_id = value.get("key_id").and_then(|v| v.as_str()).expect("key_id present");
+            assert_eq!(artifact.path, format!("artifacts/keys/{key_id}.key.enc"));
+            let material = value
+                .get("encrypted_key_material")
+                .and_then(|v| v.as_str())
+                .expect("material present");
+            assert!(!material.is_empty());
+        }
+    }
+
+    #[tokio::test]
+    async fn tampered_and_truncated_bundles_fail_closed() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+        let artifact = &manifest.artifacts[0];
+        let artifact_file = destination.join(&artifact.path);
+        let original_artifact = std::fs::read(&artifact_file).expect("artifact bytes");
+
+        // Tampered artifact byte: digest verification rejects it.
+        let mut tampered = original_artifact.clone();
+        let last = tampered.len() - 1;
+        tampered[last] ^= 0x01;
+        std::fs::write(&artifact_file, &tampered).expect("write tampered");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+            .await
+            .expect_err("tampered artifact must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+
+        // Truncated artifact: typed truncation error.
+        std::fs::write(&artifact_file, &original_artifact[..original_artifact.len() - 4]).expect("truncate");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+            .await
+            .expect_err("truncated artifact must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Truncated { .. })), "got {error:?}");
+        std::fs::write(&artifact_file, &original_artifact).expect("restore artifact");
+
+        // Tampered manifest (generation flip): sealed digest mismatch.
+        let manifest_file = destination.join(LOCAL_BUNDLE_MANIFEST_FILE);
+        let original_manifest = std::fs::read(&manifest_file).expect("manifest bytes");
+        let tampered_manifest = String::from_utf8(original_manifest.clone())
+            .expect("manifest is utf-8")
+            .replace("\"snapshot_generation\":7", "\"snapshot_generation\":8");
+        assert_ne!(tampered_manifest.as_bytes(), original_manifest.as_slice(), "tamper must apply");
+        std::fs::write(&manifest_file, tampered_manifest).expect("write tampered manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("tampered manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+
+        // Truncated manifest.
+        std::fs::write(&manifest_file, &original_manifest[..original_manifest.len() / 2]).expect("truncate manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("truncated manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Truncated { .. })), "got {error:?}");
+
+        // Missing manifest: the bundle never sealed.
+        std::fs::remove_file(&manifest_file).expect("remove manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("bundle without manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::IncompleteBundle { .. })), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn wrong_kek_is_rejected_before_decryption() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+        let artifact = &manifest.artifacts[0];
+
+        let wrong_id = BackupKek::new("other-kek", 1, [0x42; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_id)
+            .await
+            .expect_err("mismatched KEK id must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::WrongKek { .. })), "got {error:?}");
+
+        let wrong_version = BackupKek::new("backup-kek-test", 2, [0x42; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_version)
+            .await
+            .expect_err("mismatched KEK version must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::WrongKek { .. })), "got {error:?}");
+
+        // Right identity, wrong material: AEAD authentication fails closed.
+        let wrong_material = BackupKek::new("backup-kek-test", 1, [0x24; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_material)
+            .await
+            .expect_err("wrong KEK material must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn missing_salt_with_encrypted_records_fails_export() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+        fs::remove_file(client.master_key_salt_file()).await.expect("remove salt");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("export without salt must fail");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+        assert!(error.to_string().contains("salt"), "got {error}");
+    }
+
+    #[tokio::test]
+    async fn refuses_empty_key_dir_and_nonempty_destination() {
+        let (client, _key_dir) = dev_client().await;
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("empty key dir must not produce a bundle");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+
+        client.create_key("dev-key", "AES_256", None).await.expect("create key");
+        let occupied = bundle.path().join("occupied");
+        std::fs::create_dir_all(&occupied).expect("mkdir");
+        std::fs::write(occupied.join("stale"), b"leftover").expect("occupy");
+        let error = export_local_backup(&client, &test_kek(), &export_request(occupied))
+            .await
+            .expect_err("non-empty destination must be refused");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn legacy_records_export_verbatim_with_weakest_protection_label() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("modern", "AES_256", None).await.expect("create key");
+        client.create_key("legacy-key", "AES_256", None).await.expect("create key");
+
+        // Strip the protection marker to fabricate a pre-beta.9 record, the
+        // same way the local backend's own legacy-compat tests do.
+        let legacy_path = client.key_directory().join("legacy-key.key");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&legacy_path).await.expect("record")).expect("record parses");
+        record
+            .as_object_mut()
+            .expect("record is an object")
+            .remove("at_rest_protection");
+        let legacy_bytes = serde_json::to_vec_pretty(&record).expect("record serializes");
+        fs::write(&legacy_path, &legacy_bytes).await.expect("write legacy record");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::LegacyUnspecified);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(
+            kdf.protection_modes,
+            vec![AtRestProtection::EncryptedMasterKey, AtRestProtection::LegacyUnspecified]
+        );
+
+        let legacy_artifact = manifest
+            .artifacts
+            .iter()
+            .find(|a| a.path == "artifacts/keys/legacy-key.key.enc")
+            .expect("legacy artifact");
+        let decrypted = decrypt_bundle_artifact(&destination, &manifest, legacy_artifact, &kek)
+            .await
+            .expect("legacy artifact decrypts");
+        assert_eq!(decrypted.as_slice(), legacy_bytes.as_slice(), "legacy record must travel verbatim");
+    }
+
+    #[tokio::test]
+    async fn record_identity_mismatch_aborts_export() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("good", "AES_256", None).await.expect("create key");
+        std::fs::copy(client.key_directory().join("good.key"), client.key_directory().join("evil.key"))
+            .expect("plant mismatched record");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("identity mismatch must abort the export");
+        assert!(matches!(error, KmsError::InvalidKey { .. }), "got {error:?}");
+    }
+}

crates/kms/src/backup/manifest.rs

@@ -355,8 +355,10 @@ struct ManifestProbe {
 ///
 /// The manifest is the authoritative description of one backup bundle: what
 /// was captured, under which snapshot generation, protected by which backup
-/// KEK, and which restore responsibility applies. Field order is part of the
-/// canonical digest form and is frozen for this format version.
+/// KEK, and which restore responsibility applies. The digest's canonical
+/// form is the manifest's JSON value with the digest hex emptied (see
+/// [`Self::compute_digest`]); decoders verify it against the raw stored
+/// bytes and never re-serialize parsed fields.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct BackupManifest {
@@ -415,8 +417,14 @@ impl BackupManifest {
     ///
     /// Fail-closed order: truncated or malformed input, then unknown format
     /// version, then a missing completeness marker, then schema decoding
-    /// (unknown fields, duplicate fields, missing fields), then semantic
-    /// validation including digest verification.
+    /// (unknown fields, duplicate fields, missing fields), then digest
+    /// verification against the raw input bytes, then semantic validation.
+    ///
+    /// The digest is verified against the bytes as stored — parsed typed
+    /// fields are never re-serialized for verification, so a field whose
+    /// string form does not round-trip byte-identically through its parsed
+    /// representation (timestamps in environment-dependent time zone
+    /// spellings, for example) cannot produce a spurious mismatch.
     pub fn decode(bytes: &[u8]) -> Result<Self, BackupError> {
         let probe: ManifestProbe = serde_json::from_slice(bytes).map_err(map_serde_error)?;
         if probe.format_version != Self::FORMAT_VERSION {
@@ -429,7 +437,9 @@ impl BackupManifest {
             return Err(BackupError::incomplete_bundle("manifest has no completeness marker"));
         }
         let manifest: Self = serde_json::from_slice(bytes).map_err(map_serde_error)?;
-        manifest.validate()?;
+        manifest.validate_pre_digest()?;
+        Self::verify_digest_in_bytes(bytes, &manifest.manifest_digest)?;
+        manifest.validate_content()?;
         Ok(manifest)
     }
 
@@ -449,23 +459,30 @@ impl BackupManifest {
         Ok(self)
     }
 
-    /// Compute the digest over the canonical manifest bytes.
+    /// Compute the digest over the canonical manifest form.
     ///
-    /// Canonical form: compact JSON serialization of this manifest with the
-    /// digest hex emptied. Field order is struct declaration order and is
-    /// frozen for format version 1, so the same manifest content always
-    /// hashes to the same value.
+    /// Canonical form: the JSON *value* of the manifest with the digest hex
+    /// emptied, object keys rebuilt in bytewise-sorted order at every level
+    /// (see [`canonicalize_value`]), then serialized compactly. The value
+    /// layer is what makes sealing and decoding agree byte-for-byte — a
+    /// decoder recovers the identical value from the raw stored bytes
+    /// without round-tripping any typed field through parse-and-reprint —
+    /// and the explicit key sort makes the bytes independent of
+    /// `serde_json`'s map implementation (`preserve_order` on or off).
     pub fn compute_digest(&self) -> Result<ContentDigest, BackupError> {
         let mut unsealed = self.clone();
         unsealed.manifest_digest = ContentDigest::placeholder(self.manifest_digest.algorithm);
-        let canonical = serde_json::to_vec(&unsealed)
+        let value = serde_json::to_value(&unsealed)
             .map_err(|error| BackupError::corrupted(format!("manifest canonicalization failed: {error}")))?;
-        match self.manifest_digest.algorithm {
-            DigestAlgorithm::Sha256 => Ok(ContentDigest::sha256_of(&canonical)),
-        }
+        Self::digest_of_canonical_value(value, self.manifest_digest.algorithm)
     }
 
-    /// Verify the sealed digest against the current manifest content.
+    /// Verify the sealed digest against the current in-memory content.
+    ///
+    /// This is the producer-side check (sealing and [`Self::encode`]).
+    /// Decoders must use the raw stored bytes instead (see [`Self::decode`]):
+    /// re-serializing parsed fields is not guaranteed to reproduce the
+    /// stored spelling byte-for-byte.
     pub fn verify_digest(&self) -> Result<(), BackupError> {
         if !self.manifest_digest.is_well_formed() {
             return Err(BackupError::corrupted("manifest digest is not a well-formed digest value"));
@@ -478,6 +495,33 @@ impl BackupManifest {
         Ok(())
     }
 
+    /// Verify a declared digest against raw manifest bytes, normalizing only
+    /// through the JSON value layer and emptying the digest slot in place.
+    fn verify_digest_in_bytes(bytes: &[u8], declared: &ContentDigest) -> Result<(), BackupError> {
+        if !declared.is_well_formed() {
+            return Err(BackupError::corrupted("manifest digest is not a well-formed digest value"));
+        }
+        let mut value: serde_json::Value = serde_json::from_slice(bytes).map_err(map_serde_error)?;
+        let Some(slot) = value.get_mut("manifest_digest").and_then(|digest| digest.get_mut("hex")) else {
+            return Err(BackupError::corrupted("manifest has no digest slot"));
+        };
+        *slot = serde_json::Value::String(String::new());
+        if Self::digest_of_canonical_value(value, declared.algorithm)? != *declared {
+            return Err(BackupError::corrupted(
+                "manifest digest mismatch: content does not match the sealed digest",
+            ));
+        }
+        Ok(())
+    }
+
+    fn digest_of_canonical_value(value: serde_json::Value, algorithm: DigestAlgorithm) -> Result<ContentDigest, BackupError> {
+        let canonical = serde_json::to_vec(&canonicalize_value(value))
+            .map_err(|error| BackupError::corrupted(format!("manifest canonicalization failed: {error}")))?;
+        match algorithm {
+            DigestAlgorithm::Sha256 => Ok(ContentDigest::sha256_of(&canonical)),
+        }
+    }
+
     /// Look up a required artifact by kind, failing closed when absent.
     pub fn require_artifact(&self, kind: ArtifactKind) -> Result<&ArtifactDescriptor, BackupError> {
         self.artifacts
@@ -486,11 +530,21 @@ impl BackupManifest {
             .ok_or_else(|| BackupError::missing_artifact(artifact_kind_name(kind)))
     }
 
-    /// Validate the full manifest contract.
+    /// Validate the full manifest contract against the in-memory content.
     ///
-    /// This is decode-side validation and also guards [`Self::encode`], so a
-    /// producer cannot publish a manifest a decoder would reject.
+    /// This guards [`Self::encode`], so a producer cannot publish a manifest
+    /// a decoder would reject. [`Self::decode`] runs the same checks but
+    /// verifies the digest against the raw input bytes instead.
     pub fn validate(&self) -> Result<(), BackupError> {
+        self.validate_pre_digest()?;
+        self.verify_digest()?;
+        self.validate_content()
+    }
+
+    /// Checks that must run before any digest verification: an unknown
+    /// version or an unsealed bundle is reported as its own typed error, not
+    /// as a digest mismatch.
+    fn validate_pre_digest(&self) -> Result<(), BackupError> {
         if self.format_version != Self::FORMAT_VERSION {
             return Err(BackupError::UnknownVersion {
                 found: self.format_version,
@@ -500,7 +554,12 @@ impl BackupManifest {
         if self.completeness != CompletenessState::Complete {
             return Err(BackupError::incomplete_bundle("completeness marker records an in-progress bundle"));
         }
-        self.verify_digest()?;
+        Ok(())
+    }
+
+    /// Semantic validation of everything except version, completeness, and
+    /// digest integrity.
+    fn validate_content(&self) -> Result<(), BackupError> {
         require_non_empty("backup_id", &self.backup_id)?;
         require_non_empty("rustfs_version", &self.rustfs_version)?;
         require_non_empty("deployment_identity", &self.deployment_identity)?;
@@ -645,6 +704,29 @@ fn map_serde_error(error: serde_json::Error) -> BackupError {
     }
 }
 
+/// Rebuild a JSON value with object keys in bytewise-sorted order at every
+/// nesting level (array element order is preserved).
+///
+/// `serde_json`'s map keeps keys sorted by default but preserves insertion
+/// order when the `preserve_order` feature is unified into the build by any
+/// other crate. Digest bytes must not depend on that, so the ordering is
+/// imposed explicitly here instead of being inherited from the map type.
+fn canonicalize_value(value: serde_json::Value) -> serde_json::Value {
+    match value {
+        serde_json::Value::Object(map) => {
+            let mut entries: Vec<(String, serde_json::Value)> = map.into_iter().collect();
+            entries.sort_by(|a, b| a.0.cmp(&b.0));
+            let mut sorted = serde_json::Map::with_capacity(entries.len());
+            for (key, entry) in entries {
+                sorted.insert(key, canonicalize_value(entry));
+            }
+            serde_json::Value::Object(sorted)
+        }
+        serde_json::Value::Array(items) => serde_json::Value::Array(items.into_iter().map(canonicalize_value).collect()),
+        other => other,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -736,7 +818,7 @@ mod tests {
     /// canonical form and frozen. If serialization layout or field order
     /// changes, this value changes and the fixture test fails — which is the
     /// point: that is a format-version bump, not a patch.
-    const FIXTURE_DIGEST_HEX: &str = "01accb3e2bc51e12d17d1efc52ad6ab9441c50bec52c3fb29e0a9f92b725cdaa";
+    const FIXTURE_DIGEST_HEX: &str = "a8b104d61c358cd75cc9a7691d2f7d2d96f73c53653bef8940a49e96dbdf7775";
 
     fn fixture() -> String {
         FIXTURE.replace("SEALED_DIGEST_HEX", FIXTURE_DIGEST_HEX)
@@ -1010,12 +1092,39 @@ mod tests {
         let with_discovery = fixture().replace("\"completeness\"", "\"capability_discovery\": [1], \"completeness\"");
         expect_corrupted(BackupManifest::decode(with_discovery.as_bytes()), "reserved");
 
-        // Explicit null carries no data and is tolerated as absence; digest
-        // verification still passes because null slots are skipped on
-        // serialization.
+        // An explicit null slot decodes as absence at the schema layer, but
+        // sealed bundles never contain the key (`skip_serializing_if`), so
+        // inserting one after sealing is a byte-level modification and the
+        // raw-bytes digest check rejects it.
         let with_null = fixture().replace("\"completeness\"", "\"key_versions\": null, \"completeness\"");
-        let decoded = BackupManifest::decode(with_null.as_bytes()).expect("null reserved slot should decode");
-        assert_eq!(decoded.key_versions, None);
+        expect_corrupted(BackupManifest::decode(with_null.as_bytes()), "digest mismatch");
+    }
+
+    #[test]
+    fn digest_verification_survives_non_round_tripping_timestamp_spellings() {
+        // A legacy `created_at` spelling (no time zone annotation) parses via
+        // the compat fallback and re-serializes differently ("+00:00[UTC]"),
+        // and host-dependent zone spellings can do the same. Digest
+        // verification therefore operates on the raw stored bytes and must
+        // never re-serialize parsed fields.
+        let sealed = seal(local_manifest_unsealed());
+        let mut value = serde_json::to_value(&sealed).expect("manifest should convert to a value");
+        value["created_at"] = serde_json::Value::String("2026-07-30T00:00:00+00:00".to_string());
+        value["manifest_digest"]["hex"] = serde_json::Value::String(String::new());
+        let digest = BackupManifest::digest_of_canonical_value(value.clone(), DigestAlgorithm::Sha256)
+            .expect("canonical digest should compute");
+        value["manifest_digest"]["hex"] = serde_json::Value::String(digest.hex);
+        let bytes = serde_json::to_vec(&value).expect("manifest bytes");
+
+        let decoded = BackupManifest::decode(&bytes).expect("a non-round-tripping timestamp spelling must not break decoding");
+
+        // Precondition: the spelling really does not survive a typed
+        // round-trip — otherwise this test is vacuous.
+        let reserialized = serde_json::to_value(&decoded).expect("decoded manifest should convert to a value");
+        assert_ne!(reserialized["created_at"], value["created_at"]);
+        // Which is exactly why the producer-side (in-memory) digest check
+        // cannot be used on decoded manifests.
+        assert!(decoded.verify_digest().is_err());
     }
 
     #[test]

crates/kms/src/backup/mod.rs

@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Backup/restore contract types for KMS state.
+//! Backup/restore contracts and backup production for KMS state.
 //!
-//! This module is contract-only: it defines the versioned backup manifest,
-//! the per-backend responsibility matrix, typed failure modes, and the
-//! restore dry-run report. Nothing here is wired into handlers or backends;
-//! backup export, restore orchestration, and the admin API build on these
-//! types in follow-up changes.
+//! The contract side defines the versioned backup manifest, the per-backend
+//! responsibility matrix, typed failure modes, and the restore dry-run
+//! report. [`local_export`] implements the producer side for the Local
+//! backend as a crate-internal API; restore orchestration and the admin API
+//! build on these pieces in follow-up changes.
 //!
 //! # Bundle model
 //!
@@ -50,6 +50,7 @@
 mod capability;
 mod dry_run;
 mod error;
+pub mod local_export;
 mod manifest;
 
 pub use capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
@@ -57,6 +58,10 @@ pub use dry_run::{
     ExternalDependencyMismatch, RestoreBlocker, RestoreBlockerCode, RestoreConflict, RestoreConflictKind, RestoreDryRunReport,
 };
 pub use error::BackupError;
+pub use local_export::{
+    BackupKek, LOCAL_BUNDLE_MANIFEST_FILE, LocalBackupExportRequest, decrypt_bundle_artifact, export_local_backup,
+    read_local_bundle_manifest,
+};
 pub use manifest::{
     AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest,
     DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation, ReservedSlot,

crates/kms/src/deletion_worker.rs

@@ -182,7 +182,6 @@ impl DeletionWorker {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::backends::KmsClient as _;
     use crate::backends::local::LocalKmsBackend;
     use crate::config::KmsConfig;
     use crate::error::KmsError;

crates/kms/src/manager.rs

@@ -66,16 +66,19 @@ impl KmsManager {
     }
 
     /// Encrypt data with a master key
+    #[hotpath::measure]
     pub async fn encrypt(&self, request: EncryptRequest) -> Result<EncryptResponse> {
         self.backend.encrypt(request).await
     }
 
     /// Decrypt data with a master key
+    #[hotpath::measure]
     pub async fn decrypt(&self, request: DecryptRequest) -> Result<DecryptResponse> {
         self.backend.decrypt(request).await
     }
 
     /// Generate a data encryption key
+    #[hotpath::measure]
     pub async fn generate_data_key(&self, request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
         self.backend.generate_data_key(request).await
     }

crates/kms/tests/vault_fault_injection.rs

@@ -14,8 +14,8 @@
 
 //! Fault-injection matrix for the Vault backend operation policy.
 //!
-//! Offline cases run against locally injected transport faults (a closed
-//! port, a listener that never responds) — deterministic, no external
+//! Offline cases run against locally injected transport faults (a listener
+//! that never responds) — deterministic, no external
 //! dependencies. Real-Vault cases are `#[ignore]`d and need a dev Vault
 //! (default `http://127.0.0.1:8200`, override with `RUSTFS_KMS_VAULT_ADDR`).
 //!
@@ -33,9 +33,11 @@ use std::time::Duration;
 
 use metrics_util::MetricKind;
 use metrics_util::debugging::{DebugValue, DebuggingRecorder};
-use rustfs_kms::backends::KmsClient;
-use rustfs_kms::backends::vault::VaultKmsClient;
-use rustfs_kms::{KmsConfig, KmsError, VaultAuthMethod, VaultConfig};
+use rustfs_kms::backends::KmsBackend as KmsBackendTrait;
+use rustfs_kms::backends::vault::VaultKmsBackend;
+use rustfs_kms::{
+    BackendConfig, DescribeKeyRequest, KmsBackend as KmsBackendKind, KmsConfig, KmsError, VaultAuthMethod, VaultConfig,
+};
 
 const OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
 const ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
@@ -54,14 +56,23 @@ fn vault_config(address: &str, token: &str) -> VaultConfig {
     }
 }
 
-fn kms_config(attempt_timeout: Duration, retry_attempts: u32) -> KmsConfig {
+fn kms_config(vault_config: VaultConfig, attempt_timeout: Duration, retry_attempts: u32) -> KmsConfig {
     KmsConfig {
+        backend: KmsBackendKind::VaultKv2,
+        backend_config: BackendConfig::VaultKv2(Box::new(vault_config)),
+        allow_insecure_dev_defaults: true,
         timeout: attempt_timeout,
         retry_attempts,
         ..KmsConfig::default()
     }
 }
 
+fn describe_key_request(key_id: &str) -> DescribeKeyRequest {
+    DescribeKeyRequest {
+        key_id: key_id.to_string(),
+    }
+}
+
 type MetricEntry = (
     metrics_util::CompositeKey,
     Option<metrics::Unit>,
@@ -107,45 +118,6 @@ fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)])
         .sum()
 }
 
-/// Connection refused: connection-class failures are retried up to the
-/// configured budget, then surface as a backend error.
-#[test]
-fn connection_refused_is_retried_within_budget() {
-    // Reserve a loopback port and release it so nothing is listening there.
-    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a loopback port");
-    let address = format!("http://{}", listener.local_addr().expect("reserved port addr"));
-    drop(listener);
-
-    let snapshot = record_metrics(|| {
-        Box::pin(async move {
-            let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_secs(2), 2))
-                .await
-                .expect("client construction performs no network calls");
-            let error = client
-                .describe_key("fault-injection-refused", None)
-                .await
-                .expect_err("a refused connection must fail the operation");
-            assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
-        })
-    });
-
-    assert_eq!(
-        counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]),
-        2,
-        "both budgeted attempts must observe the refused connection"
-    );
-    assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1);
-    // The static-token login records its own success; the Vault read must not.
-    assert_eq!(
-        counter_value(
-            &snapshot,
-            OPERATIONS_TOTAL,
-            &[("operation", "vault_kv2_read_key"), ("outcome", "success")]
-        ),
-        0
-    );
-}
-
 /// Stalled connection: a server that accepts but never responds is cut off by
 /// the per-attempt timeout (either the policy timer or the equally sized HTTP
 /// client timeout, whichever fires first) instead of hanging forever.
@@ -166,11 +138,10 @@ fn stalled_connection_is_cut_off_by_the_attempt_timeout() {
                 }
             });
 
-            let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_millis(250), 1))
+            let client = VaultKmsBackend::new(kms_config(vault_config(&address, "unused"), Duration::from_millis(250), 1))
                 .await
                 .expect("client construction performs no network calls");
-            let error = client
-                .describe_key("fault-injection-stalled", None)
+            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-stalled"))
                 .await
                 .expect_err("a stalled request must be cut off by the attempt timeout");
             assert!(
@@ -201,11 +172,10 @@ fn real_vault_invalid_token_is_fatal_and_never_retried() {
     let snapshot = record_metrics(|| {
         Box::pin(async {
             let config = vault_config(&real_vault_address(), "fault-injection-invalid-token");
-            let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
+            let client = VaultKmsBackend::new(kms_config(config, Duration::from_secs(5), 3))
                 .await
                 .expect("client construction performs no network calls");
-            let error = client
-                .describe_key("fault-injection-forbidden", None)
+            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-forbidden"))
                 .await
                 .expect_err("an invalid token must be rejected");
             assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
@@ -235,11 +205,10 @@ fn real_vault_missing_key_is_resolved_in_one_attempt() {
     let snapshot = record_metrics(|| {
         Box::pin(async move {
             let config = vault_config(&real_vault_address(), &token);
-            let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
+            let client = VaultKmsBackend::new(kms_config(config, Duration::from_secs(5), 3))
                 .await
                 .expect("client construction performs no network calls");
-            let error = client
-                .describe_key("fault-injection-definitely-missing", None)
+            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-definitely-missing"))
                 .await
                 .expect_err("a missing key must resolve to key-not-found");
             assert!(matches!(error, KmsError::KeyNotFound { .. }), "got {error:?}");

crates/lifecycle/Cargo.toml

@@ -25,7 +25,35 @@ keywords = ["lifecycle", "storage", "rustfs", "Minio"]
 categories = ["web-programming", "development-tools", "filesystem"]
 documentation = "https://docs.rs/rustfs-lifecycle/latest/rustfs_lifecycle/"
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "rustfs-common/hotpath",
+    "rustfs-config/hotpath",
+    "rustfs-replication/hotpath",
+    "rustfs-storage-api/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-replication/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-replication/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 async-trait.workspace = true
 metrics.workspace = true
 rustfs-common.workspace = true

crates/lock/Cargo.toml

@@ -28,7 +28,21 @@ documentation = "https://docs.rs/rustfs-lock/latest/rustfs_lock/"
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/parking_lot",
+    "rustfs-io-metrics/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-io-metrics/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-io-metrics/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-io-metrics = { workspace = true }
 rustfs-utils = { workspace = true }
 async-trait.workspace = true

crates/lock/src/distributed_lock.rs

@@ -540,6 +540,7 @@ impl DistributedLock {
     }
 
     /// Acquire a lock and return a RAII guard
+    #[hotpath::measure]
     pub(crate) async fn acquire_guard(&self, request: &LockRequest) -> Result<Option<DistributedLockGuard>> {
         if self.clients.is_empty() {
             return Err(LockError::internal("No lock clients available"));
@@ -626,6 +627,7 @@ impl DistributedLock {
     }
 
     /// Convenience: acquire exclusive lock as a guard
+    #[hotpath::measure]
     pub async fn lock_guard(
         &self,
         resource: ObjectKey,
@@ -640,6 +642,7 @@ impl DistributedLock {
     }
 
     /// Convenience: acquire exclusive lock with expected contention logs suppressed
+    #[hotpath::measure]
     pub async fn lock_guard_quiet(
         &self,
         resource: ObjectKey,
@@ -655,6 +658,7 @@ impl DistributedLock {
     }
 
     /// Convenience: acquire shared lock as a guard
+    #[hotpath::measure]
     pub async fn rlock_guard(
         &self,
         resource: ObjectKey,

crates/log-analyzer/Cargo.toml

@@ -32,7 +32,14 @@ doctest = false
 name = "la-dump-anchors"
 path = "src/bin/la_dump_anchors.rs"
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 chrono = { workspace = true, features = ["serde"] }
 flate2 = { workspace = true }
 regex = { workspace = true }

crates/madmin/Cargo.toml

@@ -28,7 +28,14 @@ documentation = "https://docs.rs/rustfs-madmin/latest/rustfs_madmin/"
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 chrono = { workspace = true, features = ["serde"] }
 humantime.workspace = true
 hyper = { workspace = true, features = ["http2", "http1", "server"] }

crates/notify/Cargo.toml

@@ -26,9 +26,40 @@ categories = ["web-programming", "development-tools", "filesystem"]
 documentation = "https://docs.rs/rustfs-notify/latest/rustfs_notify/"
 
 [features]
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "rustfs-config/hotpath",
+    "rustfs-ecstore/hotpath",
+    "rustfs-s3-ops/hotpath",
+    "rustfs-s3-types/hotpath",
+    "rustfs-targets/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-s3-ops/hotpath-alloc",
+    "rustfs-s3-types/hotpath-alloc",
+    "rustfs-targets/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-ecstore/hotpath-cpu",
+    "rustfs-s3-ops/hotpath-cpu",
+    "rustfs-s3-types/hotpath-cpu",
+    "rustfs-targets/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
 demo-examples = []
 
 [dependencies]
+hotpath.workspace = true
 rustfs-config = { workspace = true, features = ["notify", "server-config-model"] }
 rustfs-ecstore = { workspace = true }
 rustfs-s3-types = { workspace = true }

crates/object-capacity/Cargo.toml

@@ -34,7 +34,33 @@ harness = false
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "rustfs-config/hotpath",
+    "rustfs-io-metrics/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-io-metrics/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-io-metrics/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-config = { workspace = true, features = ["constants"] }
 rustfs-io-metrics = { workspace = true }
 rustfs-utils = { workspace = true, features = ["os"] }

crates/object-capacity/src/capacity_manager.rs

@@ -1035,6 +1035,7 @@ impl HybridCapacityManager {
     ///
     /// Joiners subscribe to the watch channel *before* releasing the mutex, which guarantees
     /// they cannot miss the completion notification even if the leader finishes very quickly.
+    #[hotpath::measure]
     pub async fn refresh_or_join<F, Fut>(&self, source: DataSource, refresh_fn: F) -> Result<CapacityUpdate, String>
     where
         F: FnOnce() -> Fut,
@@ -1142,6 +1143,7 @@ impl HybridCapacityManager {
     }
 
     /// Start a background refresh if one is not already in flight.
+    #[hotpath::measure]
     pub async fn spawn_refresh_if_needed<F, Fut>(self: Arc<Self>, source: DataSource, refresh_fn: F) -> bool
     where
         F: FnOnce() -> Fut + Send + 'static,

crates/object-capacity/src/scan.rs

@@ -338,6 +338,7 @@ pub async fn select_capacity_refresh_disks(
     }
 }
 
+#[hotpath::measure]
 pub async fn refresh_capacity_with_scope(disks: Vec<CapacityDiskRef>, dirty_subset: bool) -> Result<CapacityUpdate, String> {
     let scan_started_at = Instant::now();
     let report = calculate_data_dir_used_capacity_report(&disks)

crates/object-data-cache/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["web-programming", "development-tools"]
 [lib]
 doctest = false
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "hotpath/tokio"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 bytes = { workspace = true, features = ["serde"] }
 metrics = { workspace = true }
 moka = { workspace = true, features = ["future"] }

crates/obs/Cargo.toml

@@ -27,6 +27,50 @@ documentation = "https://docs.rs/rustfs-obs/latest/rustfs_obs/"
 
 [features]
 default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/crossbeam",
+    "rustfs-audit/hotpath",
+    "rustfs-common/hotpath",
+    "rustfs-config/hotpath",
+    "rustfs-ecstore/hotpath",
+    "rustfs-iam/hotpath",
+    "rustfs-io-metrics/hotpath",
+    "rustfs-notify/hotpath",
+    "rustfs-security-governance/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-audit/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-iam/hotpath-alloc",
+    "rustfs-io-metrics/hotpath-alloc",
+    "rustfs-notify/hotpath-alloc",
+    "rustfs-security-governance/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-audit/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-ecstore/hotpath-cpu",
+    "rustfs-iam/hotpath-cpu",
+    "rustfs-io-metrics/hotpath-cpu",
+    "rustfs-notify/hotpath-cpu",
+    "rustfs-security-governance/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
 # Tokio runtime-level telemetry. Requires a `--cfg tokio_unstable` build; the
 # build script fails the compile when that flag is missing. Off by default so
 # ordinary builds neither pay for nor depend on Tokio's unstable API.
@@ -58,6 +102,7 @@ required-features = ["dial9"]
 workspace = true
 
 [dependencies]
+hotpath.workspace = true
 rustfs-audit = { workspace = true }
 rustfs-common = { workspace = true }
 rustfs-config = { workspace = true, features = ["observability"] }

crates/policy/Cargo.toml

@@ -30,9 +30,27 @@ workspace = true
 
 [features]
 default = []
-hotpath = ["hotpath/hotpath", "hotpath/reqwest-0-13"]
-hotpath-alloc = ["hotpath/hotpath-alloc"]
-hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/reqwest-0-13",
+    "rustfs-config/hotpath",
+    "rustfs-credentials/hotpath",
+    "rustfs-crypto/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-credentials/hotpath-alloc",
+    "rustfs-crypto/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-credentials/hotpath-cpu",
+    "rustfs-crypto/hotpath-cpu",
+]
 
 [dependencies]
 rustfs-credentials = { workspace = true }

crates/protocols/Cargo.toml

@@ -30,6 +30,55 @@ workspace = true
 
 [features]
 default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "rustfs-config/hotpath",
+    "rustfs-credentials/hotpath",
+    "rustfs-ecstore?/hotpath",
+    "rustfs-iam/hotpath",
+    "rustfs-keystone?/hotpath",
+    "rustfs-policy/hotpath",
+    "rustfs-rio?/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-tls-runtime?/hotpath",
+    "rustfs-trusted-proxies?/hotpath",
+    "rustfs-utils/hotpath",
+    "rustfs-test-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-credentials/hotpath-alloc",
+    "rustfs-ecstore?/hotpath-alloc",
+    "rustfs-iam/hotpath-alloc",
+    "rustfs-keystone?/hotpath-alloc",
+    "rustfs-policy/hotpath-alloc",
+    "rustfs-rio?/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-tls-runtime?/hotpath-alloc",
+    "rustfs-trusted-proxies?/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+    "rustfs-test-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-credentials/hotpath-cpu",
+    "rustfs-ecstore?/hotpath-cpu",
+    "rustfs-iam/hotpath-cpu",
+    "rustfs-keystone?/hotpath-cpu",
+    "rustfs-policy/hotpath-cpu",
+    "rustfs-rio?/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-tls-runtime?/hotpath-cpu",
+    "rustfs-trusted-proxies?/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+    "rustfs-test-utils/hotpath-cpu",
+]
 ftps = ["dep:libunftp", "dep:unftp-core", "dep:rustls", "dep:rustfs-tls-runtime", "dep:subtle"]
 swift = [
     "dep:rustfs-keystone",
@@ -62,6 +111,7 @@ webdav = ["dep:dav-server", "dep:hyper", "dep:hyper-util", "dep:http-body-util",
 sftp = ["dep:russh", "dep:russh-sftp", "dep:uuid", "dep:subtle", "dep:tokio-util", "dep:socket2"]
 
 [dependencies]
+hotpath.workspace = true
 # Core RustFS dependencies
 rustfs-iam = { workspace = true }
 rustfs-credentials = { workspace = true }

crates/protos/Cargo.toml

@@ -32,7 +32,38 @@ workspace = true
 name = "gproto"
 path = "src/main.rs"
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "rustfs-common/hotpath",
+    "rustfs-config/hotpath",
+    "rustfs-io-metrics/hotpath",
+    "rustfs-tls-runtime/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-io-metrics/hotpath-alloc",
+    "rustfs-tls-runtime/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-io-metrics/hotpath-cpu",
+    "rustfs-tls-runtime/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-common.workspace = true
 rustfs-io-metrics.workspace = true
 rustfs-config.workspace = true

crates/replication/Cargo.toml

@@ -25,7 +25,14 @@ keywords = ["replication", "storage", "rustfs", "Minio"]
 categories = ["web-programming", "development-tools", "filesystem"]
 documentation = "https://docs.rs/rustfs-replication/latest/rustfs_replication/"
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "hotpath/tokio"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 bytes = { workspace = true, features = ["serde"] }
 byteorder.workspace = true
 regex.workspace = true

crates/rio-v2/Cargo.toml

@@ -28,7 +28,32 @@ documentation = "https://docs.rs/rustfs-rio-v2/latest/rustfs_rio_v2/"
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "rustfs-rio/hotpath",
+    "rustfs-utils/hotpath",
+    "rustfs-filemeta/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-rio/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+    "rustfs-filemeta/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-rio/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+    "rustfs-filemeta/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 aes-gcm = { workspace = true, features = ["rand_core"] }
 bytes = { workspace = true, features = ["serde"] }
 chacha20poly1305.workspace = true

crates/rio/Cargo.toml

@@ -30,9 +30,32 @@ workspace = true
 
 [features]
 default = []
-hotpath = ["hotpath/hotpath", "hotpath/tokio", "hotpath/futures", "hotpath/reqwest-0-13"]
-hotpath-alloc = ["hotpath/hotpath-alloc"]
-hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/reqwest-0-13",
+    "rustfs-config/hotpath",
+    "rustfs-io-metrics/hotpath",
+    "rustfs-tls-runtime/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-io-metrics/hotpath-alloc",
+    "rustfs-tls-runtime/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-io-metrics/hotpath-cpu",
+    "rustfs-tls-runtime/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
 
 [dependencies]
 hotpath.workspace = true

crates/s3-ops/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["data-structures"]
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "rustfs-s3-types/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-s3-types/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-s3-types/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-s3-types = { workspace = true }
 
 [lib]

crates/s3-types/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["data-structures"]
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, features = ["raw_value"] }
 

crates/s3select-api/Cargo.toml

@@ -28,7 +28,37 @@ documentation = "https://docs.rs/rustfs-s3select-api/latest/rustfs_s3select_api/
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/parking_lot",
+    "rustfs-common/hotpath",
+    "rustfs-ecstore/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-test-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-test-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-ecstore/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-test-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 metrics = { workspace = true }
 async-trait.workspace = true
 bytes = { workspace = true, features = ["serde"] }

crates/s3select-query/Cargo.toml

@@ -28,7 +28,20 @@ documentation = "https://docs.rs/rustfs-s3select-query/latest/rustfs_s3select_qu
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/parking_lot",
+    "rustfs-s3select-api/hotpath",
+]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-s3select-api/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-s3select-api/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-s3select-api = { workspace = true }
 async-recursion = { workspace = true }
 async-trait.workspace = true

crates/scanner/Cargo.toml

@@ -29,7 +29,48 @@ documentation = "https://docs.rs/rustfs-scanner/latest/rustfs_scanner/"
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "rustfs-common/hotpath",
+    "rustfs-config/hotpath",
+    "rustfs-credentials/hotpath",
+    "rustfs-data-usage/hotpath",
+    "rustfs-ecstore/hotpath",
+    "rustfs-filemeta/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-credentials/hotpath-alloc",
+    "rustfs-data-usage/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-filemeta/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-credentials/hotpath-cpu",
+    "rustfs-data-usage/hotpath-cpu",
+    "rustfs-ecstore/hotpath-cpu",
+    "rustfs-filemeta/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-config = { workspace = true, features = ["server-config-model"] }
 rustfs-common = { workspace = true }
 rustfs-credentials = { workspace = true }
@@ -47,6 +88,7 @@ rmp-serde = { workspace = true }
 hmac = { workspace = true }
 sha2 = { workspace = true }
 rustfs-filemeta = { workspace = true }
+rustfs-lock.workspace = true
 tokio-util = { workspace = true, features = ["io", "compat", "rt"] }
 rustfs-ecstore = { workspace = true }
 rustfs-storage-api = { workspace = true }

crates/scanner/src/remote_scanner.rs

@@ -12,17 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(test)]
+use crate::RUSTFS_META_BUCKET;
 use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig};
 use crate::scanner_io::{
-    DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, cache_root_entry_info, current_cache_root_or_prepare,
-    scanner_cache_lock_resource, scanner_cache_lock_timeout, scanner_set_disk_inventory,
+    DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info,
+    current_cache_root_or_prepare, scanner_set_disk_inventory,
 };
 use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION;
-use crate::storage_api::scan::NamespaceLocking as _;
 use crate::{
     DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_CACHE_NAME, DataUsageCache, DataUsageCachePrepareOutcome, DataUsageCacheSource,
-    DataUsageEntryInfo, DataUsageScanPlanDigest, Disk, EcstoreError, RUSTFS_META_BUCKET, ScannerDiskExt as _, ScannerError,
-    ScannerObjectIO, StorageError, is_reserved_or_invalid_bucket, read_config, resolve_scanner_object_store_handle,
+    DataUsageEntryInfo, DataUsageScanPlanDigest, Disk, EcstoreError, ScannerDiskExt as _, ScannerError, ScannerObjectIO,
+    StorageError, is_reserved_or_invalid_bucket, read_config, resolve_scanner_object_store_handle,
 };
 use hmac::{Hmac, KeyInit, Mac};
 use rustfs_common::heal_channel::HealScanMode;
@@ -63,6 +64,7 @@ const NS_SCANNER_DISK_HEALTH_TIMEOUT: Duration = Duration::from_secs(5);
 const NS_SCANNER_VALIDATED_CYCLE_TTL: Duration = Duration::from_secs(1);
 const NS_SCANNER_MAX_REPLAY_SESSIONS: usize = 65_536;
 const NS_SCANNER_MAX_ERROR_CHARS: usize = 4096;
+const NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX: &str = "retry_bucket:";
 const NS_SCANNER_FRAME_AUTH_DOMAIN: &[u8] = b"rustfs-ns-scanner-frame-v3";
 
 pub const NS_SCANNER_MAX_REQUEST_BODY_SIZE: usize = 16 * 1024;
@@ -317,6 +319,13 @@ impl RemoteScannerServerError {
         }
     }
 
+    fn retry_bucket(message: impl Into<String>) -> Self {
+        Self {
+            scope: RemoteScannerErrorScope::Bucket,
+            error: ScannerError::Other(format!("{}{}", NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX, message.into())),
+        }
+    }
+
     fn into_frame(self) -> RemoteScannerErrorFrame {
         RemoteScannerErrorFrame {
             scope: self.scope,
@@ -336,6 +345,7 @@ struct RemoteScannerStreamError {
     error: StorageError,
     progress_fully_reported: bool,
     retire_worker: bool,
+    retry_bucket: bool,
 }
 
 impl RemoteScannerStreamError {
@@ -344,6 +354,7 @@ impl RemoteScannerStreamError {
             error,
             progress_fully_reported: false,
             retire_worker: true,
+            retry_bucket: false,
         }
     }
 
@@ -352,6 +363,7 @@ impl RemoteScannerStreamError {
             error,
             progress_fully_reported: true,
             retire_worker: true,
+            retry_bucket: false,
         }
     }
 
@@ -360,6 +372,16 @@ impl RemoteScannerStreamError {
             error,
             progress_fully_reported: true,
             retire_worker: false,
+            retry_bucket: false,
+        }
+    }
+
+    fn retry_bucket(error: StorageError) -> Self {
+        Self {
+            error,
+            progress_fully_reported: true,
+            retire_worker: false,
+            retry_bucket: true,
         }
     }
 
@@ -951,16 +973,14 @@ async fn scan_and_persist_local_bucket(
             ))
         })?;
     let cache_name = path_join_buf(&[&bucket, DATA_USAGE_CACHE_NAME]);
-    let lock_resource = scanner_cache_lock_resource(&cache_name);
-    let ns_lock = set
-        .new_ns_lock(RUSTFS_META_BUCKET, &lock_resource)
-        .await
-        .map_err(|err| RemoteScannerServerError::worker(format!("remote namespace scanner cache lock creation failed: {err}")))?;
-    let guard = ns_lock
-        .get_write_lock_quiet(scanner_cache_lock_timeout())
+    let guard = acquire_scanner_cache_locks(set.as_ref(), &cache_name, source)
         .await
         .map_err(|err| {
-            RemoteScannerServerError::worker(format!("remote namespace scanner cache lock acquisition failed: {err}"))
+            if err.is_contention() {
+                RemoteScannerServerError::retry_bucket(format!("remote namespace scanner cache lock contention: {err}"))
+            } else {
+                RemoteScannerServerError::worker(format!("remote namespace scanner cache lock acquisition failed: {err}"))
+            }
         })?;
     let mut cache = DataUsageCache::default();
     let revisions = cache.load_with_revisions(set.clone(), &cache_name).await.map_err(|err| {
@@ -1216,7 +1236,9 @@ fn finish_remote_scanner_stream(
             if !error.progress_fully_reported {
                 budget.cancel_after_unreported_remote_progress();
             }
-            let failure = if error.retire_worker {
+            let failure = if error.retry_bucket {
+                RemoteScannerFailure::retry_bucket(error.error)
+            } else if error.retire_worker {
                 RemoteScannerFailure::transport(error.error)
             } else {
                 RemoteScannerFailure::bucket(error.error)
@@ -1374,9 +1396,16 @@ where
                 return Ok(RemoteScannerOutcome::CycleAhead(required_cycle));
             }
             RemoteScannerFrameResult::Error(error_frame) => {
+                let retry_bucket = error_frame.message.starts_with(NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX);
+                let message = error_frame
+                    .message
+                    .strip_prefix(NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX)
+                    .map(str::trim_start)
+                    .unwrap_or(error_frame.message.as_str());
                 let error =
-                    StorageError::other(format!("remote namespace scanner failed: {}", limit_error_message(error_frame.message)));
+                    StorageError::other(format!("remote namespace scanner failed: {}", limit_error_message(message.to_string())));
                 return Err(match error_frame.scope {
+                    RemoteScannerErrorScope::Bucket if retry_bucket => RemoteScannerStreamError::retry_bucket(error),
                     RemoteScannerErrorScope::Bucket => RemoteScannerStreamError::bucket(error),
                     RemoteScannerErrorScope::Worker => RemoteScannerStreamError::reconciled(error),
                 });
@@ -2491,6 +2520,42 @@ mod tests {
         assert!(!budget.budget_elapsed());
     }
 
+    #[tokio::test]
+    async fn retry_bucket_error_terminal_frame_requeues_bucket_work() {
+        let request_id = Uuid::new_v4();
+        let writer_auth = FrameAuthenticator::for_test(request_id);
+        let reader_auth = FrameAuthenticator::for_test(request_id);
+        let (mut writer, reader) = tokio::io::duplex(4096);
+        tokio::spawn(async move {
+            let mut sequence = 0;
+            write_frame(
+                &mut writer,
+                &writer_auth,
+                &mut sequence,
+                &RemoteScannerFrame::terminal(
+                    RemoteScannerProgress::default(),
+                    RemoteScannerFrameResult::Error(RemoteScannerErrorFrame {
+                        scope: RemoteScannerErrorScope::Bucket,
+                        message: format!("{NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX} cache lock contention"),
+                    }),
+                ),
+            )
+            .await
+            .expect("retry-bucket error terminal frame should write");
+        });
+
+        let parent = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default());
+        let stream_result =
+            consume_remote_scanner_stream(reader, parent, budget.clone(), "bucket", TEST_SOURCE, TEST_PLAN_DIGEST, reader_auth)
+                .await;
+        let error = finish_remote_scanner_stream(stream_result, budget.as_ref()).expect_err("retry-bucket frame must fail");
+
+        assert!(error.retry_bucket_work());
+        assert!(!error.retire_worker());
+        assert!(error.to_string().contains("cache lock contention"));
+    }
+
     #[tokio::test]
     async fn worker_error_terminal_frame_retires_worker_without_cancelling_reported_budget() {
         let request_id = Uuid::new_v4();

crates/scanner/src/scanner.rs

@@ -949,7 +949,7 @@ pub(crate) async fn probe_scanner_activity(storeapi: &ECStore, distributed: bool
                 }
                 SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION => {
                     return Err(format!(
-                        "scanner activity peer {host} cannot acknowledge distributed dirty usage with protocol {}",
+                        "scanner activity peer {host} cannot safely share scanner cache locks with protocol {}",
                         SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION
                     ));
                 }
@@ -2531,6 +2531,7 @@ where
 }
 
 #[instrument(skip_all)]
+#[hotpath::measure]
 async fn run_data_scanner_cycle(
     ctx: &CancellationToken,
     storeapi: &Arc<ECStore>,
@@ -7113,9 +7114,17 @@ mod tests {
                 ..scanner_node_activity("epoch-a", 7, 3)
             },
         )]);
+        let previous = BTreeMap::from([(
+            "node-2".to_string(),
+            ScannerNodeActivity {
+                protocol_version: SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION,
+                ..scanner_node_activity("epoch-a", 7, 3)
+            },
+        )]);
         let current = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
 
         assert_ne!(scanner_activity_snapshot_digest(&legacy), scanner_activity_snapshot_digest(&current));
+        assert_ne!(scanner_activity_snapshot_digest(&previous), scanner_activity_snapshot_digest(&current));
     }
 
     #[test]

crates/scanner/src/scanner_io.rs

@@ -30,6 +30,7 @@ use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, e
 use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS};
 use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo};
 use rustfs_filemeta::FileMeta;
+use rustfs_lock::{LockError, NamespaceLockGuard};
 use rustfs_utils::path::path_join_buf;
 use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration};
 use sha2::{Digest as _, Sha256};
@@ -944,14 +945,85 @@ fn checked_bucket_usage_info(entry: &DataUsageEntry) -> Option<BucketUsageInfo>
     Some(usage)
 }
 
-pub(crate) fn scanner_cache_lock_resource(cache_name: &str) -> String {
-    path_join_buf(&[crate::BUCKET_META_PREFIX, cache_name, SCANNER_CACHE_LOCK_SUFFIX])
+pub(crate) fn scanner_cache_lock_resource(cache_name: &str, source: DataUsageCacheSource) -> String {
+    let lock_name = format!("{SCANNER_CACHE_LOCK_SUFFIX}.pool-{}.set-{}", source.pool_index, source.set_index);
+    path_join_buf(&[crate::BUCKET_META_PREFIX, cache_name, &lock_name])
 }
 
 pub(crate) fn scanner_cache_lock_timeout() -> Duration {
     Duration::from_secs(rustfs_utils::get_env_u64("RUSTFS_LOCK_ACQUIRE_TIMEOUT", 5))
 }
 
+#[derive(Debug)]
+pub(crate) struct ScannerCacheLockGuards {
+    scoped: NamespaceLockGuard,
+}
+
+impl ScannerCacheLockGuards {
+    pub(crate) fn is_lock_lost(&self) -> bool {
+        self.scoped.is_lock_lost()
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum ScannerCacheLockError {
+    Create { resource: String, source: Error },
+    Acquire { resource: String, source: LockError },
+}
+
+impl ScannerCacheLockError {
+    pub(crate) fn state(&self) -> &'static str {
+        match self {
+            Self::Create { .. } => "lock_create_failed",
+            Self::Acquire { .. } => "lock_acquire_failed",
+        }
+    }
+
+    pub(crate) fn is_contention(&self) -> bool {
+        matches!(
+            self,
+            Self::Acquire {
+                source: LockError::Timeout { .. } | LockError::AlreadyLocked { .. },
+                ..
+            }
+        )
+    }
+}
+
+impl std::fmt::Display for ScannerCacheLockError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Create { resource, source } => write!(formatter, "create scanner cache lock {resource}: {source}"),
+            Self::Acquire { resource, source } => write!(formatter, "acquire scanner cache lock {resource}: {source}"),
+        }
+    }
+}
+
+pub(crate) async fn acquire_scanner_cache_locks(
+    store: &SetDisks,
+    cache_name: &str,
+    source: DataUsageCacheSource,
+) -> std::result::Result<ScannerCacheLockGuards, ScannerCacheLockError> {
+    let timeout = scanner_cache_lock_timeout();
+    let scoped_resource = scanner_cache_lock_resource(cache_name, source);
+    let scoped_lock = store
+        .new_ns_lock(RUSTFS_META_BUCKET, &scoped_resource)
+        .await
+        .map_err(|source| ScannerCacheLockError::Create {
+            resource: scoped_resource.clone(),
+            source,
+        })?;
+    let scoped = scoped_lock
+        .get_write_lock_quiet(timeout)
+        .await
+        .map_err(|source| ScannerCacheLockError::Acquire {
+            resource: scoped_resource,
+            source,
+        })?;
+
+    Ok(ScannerCacheLockGuards { scoped })
+}
+
 async fn await_scanner_disk_shutdown<F>(scan: Pin<&mut F>)
 where
     F: Future,
@@ -1697,6 +1769,19 @@ mod publish_gate_tests {
         assert!(!cache_snapshot_is_current(&cache, "photos", source, 11, 0, second));
     }
 
+    #[test]
+    fn scanner_cache_lock_resource_is_scoped_to_cache_source() {
+        let cache_name = "photos/.usage-cache.bin";
+        let first_source = DataUsageCacheSource::new(0, 1);
+        let same_source = DataUsageCacheSource::new(0, 1);
+        let other_source = DataUsageCacheSource::new(1, 0);
+
+        let first = scanner_cache_lock_resource(cache_name, first_source);
+        assert_eq!(first, scanner_cache_lock_resource(cache_name, same_source));
+        assert_ne!(first, scanner_cache_lock_resource(cache_name, other_source));
+        assert!(first.ends_with(".scanner-cycle.lock.pool-0.set-1"));
+    }
+
     #[test]
     fn count_budget_serializes_set_and_disk_work() {
         assert_eq!(scanner_budgeted_concurrency_limit(8, true), 1);
@@ -1797,24 +1882,7 @@ async fn persist_and_publish_cache_snapshot(
     cache_cycle_floor: &AtomicU64,
 ) -> Option<SystemTime> {
     let source = cache_snapshot.info.source?;
-    let lock_resource = scanner_cache_lock_resource(DATA_USAGE_CACHE_NAME);
-    let ns_lock = match store.new_ns_lock(RUSTFS_META_BUCKET, &lock_resource).await {
-        Ok(lock) => lock,
-        Err(err) => {
-            error!(
-                target: "rustfs::scanner::io",
-                event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                component = LOG_COMPONENT_SCANNER,
-                subsystem = LOG_SUBSYSTEM_IO,
-                cache_name = DATA_USAGE_CACHE_NAME,
-                state = "lock_create_failed",
-                error = %err,
-                "Scanner cache snapshot lock creation failed"
-            );
-            return None;
-        }
-    };
-    let guard = match ns_lock.get_write_lock_quiet(scanner_cache_lock_timeout()).await {
+    let guard = match acquire_scanner_cache_locks(store.as_ref(), DATA_USAGE_CACHE_NAME, source).await {
         Ok(guard) => guard,
         Err(err) => {
             error!(
@@ -1823,7 +1891,7 @@ async fn persist_and_publish_cache_snapshot(
                 component = LOG_COMPONENT_SCANNER,
                 subsystem = LOG_SUBSYSTEM_IO,
                 cache_name = DATA_USAGE_CACHE_NAME,
-                state = "lock_acquire_failed",
+                state = err.state(),
                 error = %err,
                 "Scanner cache snapshot lock acquisition failed"
             );
@@ -3080,32 +3148,35 @@ impl ScannerIOCache for SetDisks {
                         None
                     };
 
-                    // Lock order: scanner leader fence -> per-bucket cache lock ->
-                    // cache object read/write. The cache lock stays outermost for
-                    // local and rolling-upgrade workers so leader failover cannot
-                    // execute the same bucket concurrently.
-                    let lock_resource = scanner_cache_lock_resource(&cache_name);
-                    let ns_lock = match store_clone_clone.new_ns_lock(RUSTFS_META_BUCKET, &lock_resource).await {
-                        Ok(lock) => lock,
-                        Err(e) => {
-                            record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
-                            error!(
-                                target: "rustfs::scanner::io",
-                                event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                                component = LOG_COMPONENT_SCANNER,
-                                subsystem = LOG_SUBSYSTEM_IO,
-                                bucket = %bucket.name,
-                                cache_name = %cache_name,
-                                state = "lock_create_failed",
-                                error = %e,
-                                "Scanner bucket cache lock creation failed"
-                            );
-                            continue;
-                        }
-                    };
-                    let cache_guard = match ns_lock.get_write_lock_quiet(scanner_cache_lock_timeout()).await {
+                    // Lock order: scanner leader fence -> set-scoped per-bucket cache lock ->
+                    // cache object read/write.
+                    let cache_guard = match acquire_scanner_cache_locks(store_clone_clone.as_ref(), &cache_name, source).await {
                         Ok(guard) => guard,
                         Err(e) => {
+                            if e.is_contention() {
+                                if requeue_bucket_work(&bucket_tx_clone, &bucket, &mut work_guard).await {
+                                    increment_disk_bucket_scans_queued(
+                                        &queued_disk_bucket_scans_clone,
+                                        &pool_label_clone,
+                                        &set_label_clone,
+                                    );
+                                } else {
+                                    record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
+                                }
+                                debug!(
+                                    target: "rustfs::scanner::io",
+                                    event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                                    component = LOG_COMPONENT_SCANNER,
+                                    subsystem = LOG_SUBSYSTEM_IO,
+                                    bucket = %bucket.name,
+                                    cache_name = %cache_name,
+                                    state = "lock_contention_requeued",
+                                    error = %e,
+                                    "Scanner bucket cache lock contention requeued bucket work"
+                                );
+                                break;
+                            }
+
                             record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
                             error!(
                                 target: "rustfs::scanner::io",
@@ -3114,7 +3185,7 @@ impl ScannerIOCache for SetDisks {
                                 subsystem = LOG_SUBSYSTEM_IO,
                                 bucket = %bucket.name,
                                 cache_name = %cache_name,
-                                state = "lock_acquire_failed",
+                                state = e.state(),
                                 error = %e,
                                 "Scanner bucket cache lock acquisition failed"
                             );
@@ -3894,6 +3965,52 @@ mod tests {
         (temp_dir, store)
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn scanner_cache_locks_block_same_source_workers() {
+        let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+        let set = &store.pools[0].disk_set[0];
+        let source = DataUsageCacheSource::new(0, 0);
+        let cache_name = "photos/.usage-cache.bin";
+
+        let guards = acquire_scanner_cache_locks(set.as_ref(), cache_name, source)
+            .await
+            .expect("scanner cache locks should be acquired");
+        let scoped_lock = set
+            .new_ns_lock(RUSTFS_META_BUCKET, &scanner_cache_lock_resource(cache_name, source))
+            .await
+            .expect("scoped scanner cache lock should be created");
+        let scoped_err = scoped_lock
+            .get_write_lock_quiet(Duration::from_millis(100))
+            .await
+            .expect_err("same-source workers must be blocked while scanner cache lock is held");
+        assert!(matches!(scoped_err, LockError::Timeout { .. } | LockError::AlreadyLocked { .. }));
+
+        drop(guards);
+        acquire_scanner_cache_locks(set.as_ref(), cache_name, source)
+            .await
+            .expect("scanner cache locks should be released when guards drop");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn scanner_cache_locks_allow_cross_source_workers() {
+        let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+        let first_set = &store.pools[0].disk_set[0];
+        let second_set = &store.pools[1].disk_set[0];
+        let cache_name = "photos/.usage-cache.bin";
+
+        let first = acquire_scanner_cache_locks(first_set.as_ref(), cache_name, DataUsageCacheSource::new(0, 0))
+            .await
+            .expect("first source scanner cache locks should be acquired");
+        let second = acquire_scanner_cache_locks(second_set.as_ref(), cache_name, DataUsageCacheSource::new(1, 0))
+            .await
+            .expect("different source scanner cache locks should not contend");
+
+        assert!(!first.is_lock_lost());
+        assert!(!second.is_lock_lost());
+    }
+
     #[tokio::test]
     async fn data_usage_publish_fails_when_receiver_is_closed() {
         let (updates, receiver) = mpsc::channel(1);

crates/security-governance/Cargo.toml

@@ -30,5 +30,12 @@ doctest = false
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 thiserror = { workspace = true }

crates/signer/Cargo.toml

@@ -25,7 +25,14 @@ keywords = ["digital-signature", "verification", "integrity", "rustfs", "Minio"]
 categories = ["web-programming", "development-tools", "cryptography"]
 documentation = "https://docs.rs/rustfs-signer/latest/rustfs_signer/"
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "rustfs-utils/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 tracing.workspace = true
 bytes = { workspace = true, features = ["serde"] }
 http.workspace = true

crates/storage-api/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["web-programming", "development-tools", "filesystem"]
 [lib]
 doctest = false
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-filemeta/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-filemeta/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-filemeta/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 async-trait.workspace = true
 # Storage-facing replication contracts are isolated in src/replication.rs until
 # the underlying wire types can move without creating a replication/storage-api cycle.

crates/storage-api/src/lib.rs

@@ -28,8 +28,8 @@ pub const NS_SCANNER_SESSION_SEQUENCE_QUERY: &str = "ns_scanner_session_sequence
 pub const NS_SCANNER_PROTOCOL_VERSION_QUERY: &str = "ns_scanner_protocol";
 pub const NS_SCANNER_PROTOCOL_VERSION: u16 = 3;
 pub const SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION: u32 = 0;
-pub const SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION: u32 = 4;
-pub const SCANNER_ACTIVITY_PROTOCOL_VERSION: u32 = 5;
+pub const SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION: u32 = 5;
+pub const SCANNER_ACTIVITY_PROTOCOL_VERSION: u32 = 6;
 
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(deny_unknown_fields)]

crates/targets/Cargo.toml

@@ -11,7 +11,41 @@ keywords = ["file-system", "notification", "target", "rustfs", "Minio"]
 categories = ["web-programming", "development-tools", "filesystem"]
 documentation = "https://docs.rs/rustfs-targets/latest/rustfs_targets/"
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/futures",
+    "hotpath/parking_lot",
+    "hotpath/reqwest-0-13",
+    "rustfs-config/hotpath",
+    "rustfs-extension-schema/hotpath",
+    "rustfs-s3-types/hotpath",
+    "rustfs-tls-runtime/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-extension-schema/hotpath-alloc",
+    "rustfs-s3-types/hotpath-alloc",
+    "rustfs-tls-runtime/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-extension-schema/hotpath-cpu",
+    "rustfs-s3-types/hotpath-cpu",
+    "rustfs-tls-runtime/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-config = { workspace = true, features = ["notify", "audit", "server-config-model"] }
 rustfs-extension-schema = { workspace = true }
 rustfs-tls-runtime = { workspace = true }

crates/targets/src/runtime/mod.rs

@@ -522,6 +522,7 @@ fn snapshot_from_delivery(target_id: TargetID, delivery: TargetDeliverySnapshot)
     }
 }
 
+#[hotpath::measure]
 pub async fn init_target_and_optionally_start_replay<E, F, G>(
     target: Box<dyn Target<E> + Send + Sync>,
     on_replay_start: F,
@@ -557,6 +558,7 @@ where
     Some((shared, cancel))
 }
 
+#[hotpath::measure]
 pub(crate) async fn prepare_target<E>(
     target: Box<dyn Target<E> + Send + Sync>,
     cancellation: Option<&CancellationToken>,
@@ -598,6 +600,7 @@ where
 
 type ActivatedTarget<E> = (SharedTarget<E>, Option<(mpsc::Sender<()>, JoinHandle<()>)>);
 
+#[hotpath::measure]
 pub async fn activate_targets_with_replay<E, F, Fut>(
     targets: Vec<Box<dyn Target<E> + Send + Sync>>,
     mut activate_one: F,
@@ -670,6 +673,7 @@ fn seed_interval_start(now: tokio::time::Instant, interval: Duration) -> tokio::
     now.checked_sub(interval).unwrap_or(now)
 }
 
+#[hotpath::measure]
 async fn stream_replay_worker<E>(
     store: &mut (dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send),
     target: SharedTarget<E>,
@@ -805,6 +809,7 @@ async fn stream_replay_worker<E>(
 /// Returns `true` if a cancel signal was observed while processing (e.g. during
 /// retry backoff), so the caller can stop promptly instead of continuing to
 /// drain a store that a replacement worker may already own.
+#[hotpath::measure]
 async fn process_replay_batch<E>(
     store: &(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send),
     batch_keys: &mut Vec<Key>,

crates/targets/src/target/webhook.rs

@@ -85,6 +85,7 @@ fn classify_probe_error(err: &reqwest::Error) -> TargetHealthReason {
     TargetHealthReason::Unreachable
 }
 
+#[hotpath::measure]
 async fn probe_health_url(client: &Client, health_check_url: &Url) -> TargetHealth {
     match tokio::time::timeout(WEBHOOK_HEALTH_TIMEOUT, client.head(health_check_url.as_str()).send()).await {
         Ok(Ok(_)) => TargetHealth::online(TargetHealthReason::Reachable),
@@ -480,6 +481,7 @@ where
         build_queued_payload(event)
     }
 
+    #[hotpath::measure]
     async fn send_body(&self, body: Vec<u8>, meta: &QueuedPayloadMeta) -> Result<(), TargetError> {
         debug!(
             event = EVENT_WEBHOOK_DELIVERY_STATE,

crates/test-utils/Cargo.toml

@@ -25,7 +25,32 @@ keywords = ["testing", "storage", "rustfs", "Minio"]
 categories = ["development-tools", "filesystem"]
 documentation = "https://docs.rs/rustfs-test-utils/latest/rustfs_test_utils/"
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "rustfs-ecstore/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-data-usage/hotpath",
+]
+hotpath-alloc = [
+    "hotpath",
+    "hotpath/hotpath-alloc",
+    "rustfs-ecstore/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-data-usage/hotpath-alloc",
+]
+hotpath-cpu = [
+    "hotpath",
+    "hotpath/hotpath-cpu",
+    "rustfs-ecstore/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-data-usage/hotpath-cpu",
+]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-ecstore = { workspace = true }
 rustfs-storage-api = { workspace = true }
 tokio = { workspace = true, features = ["fs", "rt-multi-thread"] }

crates/tls-runtime/Cargo.toml

@@ -27,7 +27,14 @@ categories = ["network-programming", "web-programming", "development-tools"]
 [lints]
 workspace = true
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-common/hotpath", "rustfs-config/hotpath"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-common/hotpath-alloc", "rustfs-config/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-common/hotpath-cpu", "rustfs-config/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 rustfs-common.workspace = true
 rustfs-config.workspace = true
 arc-swap.workspace = true

crates/trusted-proxies/Cargo.toml

@@ -24,7 +24,20 @@ description = " RustFS Trusted Proxies module provides secure and efficient mana
 keywords = ["trusted-proxies", "network-security", "rustfs", "proxy-management"]
 categories = ["network-programming", "security", "web-programming"]
 
+[features]
+default = []
+hotpath = [
+    "hotpath/hotpath",
+    "hotpath/tokio",
+    "hotpath/reqwest-0-13",
+    "rustfs-config/hotpath",
+    "rustfs-utils/hotpath",
+]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-config/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-config/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 async-trait = { workspace = true }
 axum = { workspace = true }
 http = { workspace = true }

crates/trusted-proxies/src/cloud/ranges.rs

@@ -227,6 +227,7 @@ pub struct GoogleCloudIpRanges;
 
 impl GoogleCloudIpRanges {
     /// Fetches the latest Google Cloud IP ranges from their official source.
+    #[hotpath::measure]
     pub async fn fetch() -> Result<Vec<IpNetwork>, AppError> {
         let client = Client::builder()
             .timeout(Duration::from_secs(10))

crates/utils/Cargo.toml

@@ -25,6 +25,7 @@ keywords = ["utilities", "hashing", "compression", "network", "rustfs"]
 categories = ["web-programming", "development-tools", "cryptography"]
 
 [dependencies]
+hotpath.workspace = true
 base64-simd = { workspace = true, optional = true }
 blake2 = { workspace = true, optional = true }
 brotli = { workspace = true, optional = true }
@@ -76,6 +77,9 @@ workspace = true
 
 [features]
 default = ["ip"]  # features that are enabled by default
+hotpath = ["hotpath/hotpath", "hotpath/tokio", "hotpath/futures", "hotpath/reqwest-0-13"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 ip = ["dep:local-ip-address"]  # ip characteristics and their dependencies
 net = ["ip", "dep:url", "dep:netif", "dep:futures", "dep:transform-stream", "dep:bytes", "dep:hyper", "dep:tokio"]  # network features with DNS resolver
 egress = ["ip", "dep:reqwest", "dep:tokio", "dep:url"]

crates/zip/Cargo.toml

@@ -32,7 +32,14 @@ doctest = false
 name = "zip_benchmark"
 harness = false
 
+[features]
+default = []
+hotpath = ["hotpath/hotpath", "hotpath/tokio"]
+hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
+
 [dependencies]
+hotpath.workspace = true
 async-compression = { workspace = true, features = [
     "tokio",
     "bzip2",

docs/architecture/compat-cleanup-register.md

@@ -17,7 +17,7 @@ for later deletion.
 - `rustfs-5416-kubernetes-alias-dns` Kubernetes endpoint identity fallback: deployments created before explicit local endpoint identity may use resolvable aliases that do not match the Pod hostname. An implicit auto-mode zero match retains legacy DNS locality with a bounded deadline, while ambiguous matches and invalid explicit anchors still fail closed. Remove the fallback after every supported direct-upgrade chart and deployment manifest provides a canonical RUSTFS_LOCAL_ENDPOINT_HOST for domain-based distributed topologies.
 - `rustfs-5416-zero-retry-delay` startup retry-delay validation: releases before bounded topology convergence accept RUSTFS_STARTUP_TOPOLOGY_RETRY_MAX_DELAY values of 0 or 0ms. New servers replace those values with the safe nonzero default so a direct upgrade neither fails startup nor enters a busy loop. Reject zero after the minimum supported direct-upgrade release validates or rewrites this setting before rollout.
 - `scanner-usage-v2` persisted scanner usage migration: pre-v2 scanners write `.usage.json`, so upgraded clusters read that primary/backup pair only while `.usage.v2.json` is absent and continue removing deleted buckets from legacy copies that still exist. The additive usage_snapshot_complete field in `.usage.v2.json` must remain optional while mixed-version clusters are supported; a missing field means the snapshot is not authoritative. Remove the legacy object fallback and cleanup only after every supported direct-upgrade source writes `.usage.v2.json`.
-- `ns-scanner-rpc-v3` namespace scanner capability and activity handshake: old peers and legacy internode transports lack the authenticated startup-epoch handshake. The oldest peers send an empty activity request and receive a field-empty protocol-0 response. Protocol v4 binds the challenge and response topology but cannot authenticate distributed dirty-usage state. Current protocol v5 binds the request version, acknowledgement target and generation, and the response dirty-usage state. Servers retain protocol-0 and protocol-v4 codecs for rolling upgrades, while the distributed scanner publishes usage only after every peer returns authenticated protocol v5 state. Scanner selection treats HTTP 404/405/426 and the legacy MethodNotAllowed default as an explicit lack of remote scanner v3 support and assigns those disks to coordinator-driven workers; transient capability failures remain incomplete and do not activate the fallback. Remove the coordinator fallback after the minimum supported RustFS peer version implements namespace scanner protocol v3, remove protocol-0 activity requests and responses after every supported peer implements authenticated scanner activity protocol v4, and remove the protocol-v4 activity codec after every supported peer implements protocol v5; future protocol revisions must keep the same dual-version server/codec window before changing the advertised version.
+- `ns-scanner-rpc-v3` namespace scanner capability and activity handshake: old peers and legacy internode transports lack the authenticated startup-epoch handshake. The oldest peers send an empty activity request and receive a field-empty protocol-0 response. Protocol v4 binds the challenge and response topology but cannot authenticate distributed dirty-usage state. Protocol v5 binds the request version, acknowledgement target and generation, and the response dirty-usage state, but predates set-scoped scanner cache locks. Current protocol v6 additionally fences scanner cache lock-domain changes, so distributed scanner cycles publish usage only after every peer reports protocol v6 state. Servers retain protocol-0 and protocol-v4 codecs for rolling upgrades, while protocol-v5 peers are treated as previous-version peers that cannot safely participate in the new cache lock domain. Scanner selection treats HTTP 404/405/426 and the legacy MethodNotAllowed default as an explicit lack of remote scanner v3 support and assigns those disks to coordinator-driven workers; transient capability failures remain incomplete and do not activate the fallback. Remove the coordinator fallback after the minimum supported RustFS peer version implements namespace scanner protocol v3, remove protocol-0 activity requests and responses after every supported peer implements authenticated scanner activity protocol v4, remove the protocol-v4 activity codec after every supported peer implements protocol v5, and remove protocol-v5 previous-version rejection after every supported peer implements protocol v6; future protocol revisions must keep the same dual-version server/codec window before changing the advertised version.
 - `#4648` walk-dir stream completion capability: old clients can append fallback output to an already-used metacache writer after a terminal body error, so servers emit terminal walk errors only to clients that sign the `walk_dir_stream_completion=error-v1` query capability and its request-body digest. Remove the legacy clean-EOF path after the minimum supported RustFS peer version always advertises this capability.
 - `heal-rpc-auth-v2` internode gRPC authentication: servers temporarily accept legacy prefix signatures so old peers remain available during rolling upgrades. Remove the legacy fallback after the minimum supported RustFS peer version sends v2 authentication on every internode gRPC request.
 - `disk-mutation-body-digest` internode mutating disk RPCs: servers temporarily accept mutating disk RPCs (RenameData, DeleteVersion, DeleteVersions, WriteMetadata, UpdateMetadata, WriteAll, Delete, DeletePaths, RenameFile, RenamePart, DeleteVolume, MakeVolume, MakeVolumes) that carry no signature-bound canonical body digest, so peers from releases that predate body-digest signing remain available during rolling upgrades. Accepted digestless mutations increment the internode body-digest fallback counter; that counter must read zero fleet-wide across a release window before RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT is enabled. Because body-bound requests now consume replay-cache nonces on the receiver, deploy the raised RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY default fleet-wide before enabling strict mode, and watch the internode replay-cache overflow counter for undersized capacity during the rollout. Remove the digestless fallback after the minimum supported RustFS peer version body-binds every mutating disk RPC.

docs/operations/hotpath-warp-abba-runbook.md

@@ -0,0 +1,277 @@
+# Hotpath warp ABBA validation runbook
+
+This runbook describes how to collect formal Linux or production-cluster
+evidence for hotpath performance changes. Use it when a short local A/B smoke
+run is too noisy to decide whether a regression is real.
+
+The ABBA runner executes each workload and drive-sync cell as:
+
+```text
+A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline
+```
+
+`B1` and `B2` are compared with `A1` to measure the candidate delta. `A2` is
+also compared with `A1` to measure baseline drift. Treat a candidate regression
+as actionable only when the `A2` drift is passing or materially smaller than
+the `B1` and `B2` delta for the same workload.
+
+## Scope
+
+Use this runbook for hotpath profiling and performance validation of RustFS
+object I/O changes, especially when CPU, memory allocation, lock/channel wait
+time, request throughput, or tail latency is the review question.
+
+The script validates the same workload matrix as the hotpath warp A/B gate:
+
+| Workload | mode | size |
+| --- | --- | --- |
+| `put-4kib` | put | 4KiB |
+| `put-4mib` | put | 4MiB |
+| `get-4kib` | get | 4KiB |
+| `get-4mib` | get | 4MiB |
+| `get-10mib` | get | 10MiB |
+| `mixed-256k` | mixed | 256KiB |
+
+Each workload runs with `RUSTFS_DRIVE_SYNC_ENABLE=true` and
+`RUSTFS_DRIVE_SYNC_ENABLE=false`, so a full ABBA pass produces 48 measurement
+cells: 6 workloads x 2 drive-sync modes x 4 ABBA legs.
+
+## Prerequisites
+
+Run the formal pass on Linux, not on a laptop smoke environment.
+
+Required tools on the bench host:
+
+- `bash`, `curl`, `git`, and core GNU userland.
+- `warp` on `PATH`, or pass `--warp-bin`.
+- Two RustFS Linux binaries: one baseline and one candidate.
+- Enough isolated disks or directories for the local runner, or an externally
+  managed RustFS cluster for production-like validation.
+- Stable host telemetry collection such as `pidstat`, `mpstat`, `iostat`,
+  `sar`, `perf`, `heaptrack`, or the platform's equivalent observability stack.
+
+Cluster-mode requirements:
+
+- A deploy hook that can replace the RustFS binary on every node.
+- The hook must apply `RUSTFS_DRIVE_SYNC_ENABLE` for the current ABBA leg.
+- The hook must restart RustFS and return only after the rollout command has
+  been accepted. The ABBA script performs the HTTP readiness wait.
+- The benchmark client should run outside the RustFS nodes when possible.
+- Do not run against a production data set unless the workload bucket and test
+  credentials are isolated and approved for destructive benchmark traffic.
+
+## Build the binaries
+
+Build the baseline from the comparison commit, usually `origin/main` or the
+previous accepted release:
+
+```bash
+git fetch origin main
+git switch --detach origin/main
+cargo build --release -p rustfs --bins
+cp target/release/rustfs /tmp/rustfs-baseline
+```
+
+Build the candidate from the PR commit:
+
+```bash
+git switch <candidate-branch>
+cargo build --release -p rustfs --bins
+cp target/release/rustfs /tmp/rustfs-candidate
+```
+
+For cross-compiled cluster binaries, keep both outputs on the bench host and
+make the deploy hook copy the selected binary to the cluster. The ABBA runner
+passes the selected binary path through `HOTPATH_ABBA_BINARY`.
+
+## Local Linux runner
+
+Use local mode for a dedicated Linux runner with disposable data paths. This is
+not a substitute for a production-like cluster, but it is useful before spending
+cluster time.
+
+```bash
+scripts/run_hotpath_warp_abba.sh \
+  --baseline-bin /tmp/rustfs-baseline \
+  --candidate-bin /tmp/rustfs-candidate \
+  --address 127.0.0.1:9000 \
+  --data-root /var/tmp/rustfs-hotpath-abba \
+  --disks 4 \
+  --duration 120s \
+  --rounds 3 \
+  --cooldown 30 \
+  --concurrency 16 \
+  --out-dir target/hotpath-abba/linux-local
+```
+
+The script starts and stops RustFS for each ABBA leg. The data root is
+throwaway and should not contain important data.
+
+## Production-like cluster runner
+
+Use external mode when RustFS lifecycle is managed by ansible, systemd, a
+cluster scheduler, or a dedicated deployment harness. In this mode the ABBA
+script does not start RustFS directly; it calls `--deploy-hook` before each leg
+and then waits for `http://<endpoint><health-path>`.
+
+The deploy hook receives:
+
+| Environment variable | Value |
+| --- | --- |
+| `HOTPATH_ABBA_LEG` | `A1`, `B1`, `B2`, or `A2` |
+| `HOTPATH_ABBA_PHASE` | `baseline` or `candidate` |
+| `HOTPATH_ABBA_BINARY` | selected baseline or candidate binary path |
+| `HOTPATH_ABBA_DRIVE_SYNC` | `true` or `false` |
+
+Example ansible-shaped command:
+
+```bash
+scripts/run_hotpath_warp_abba.sh \
+  --baseline-bin /srv/rustfs-binaries/rustfs-baseline \
+  --candidate-bin /srv/rustfs-binaries/rustfs-candidate \
+  --endpoint rustfs-bench.example.internal:9000 \
+  --deploy-hook '
+    set -euo pipefail
+    cd /srv/rustfs-ansible
+    cp "${HOTPATH_ABBA_BINARY:?}" roles/rustfs/files/rustfs
+    export RUSTFS_DRIVE_SYNC_ENABLE="${HOTPATH_ABBA_DRIVE_SYNC:?}"
+    ansible-playbook -f 4 -l bench rustfs-manage.yml --tags stop
+    ansible-playbook -f 4 -l bench rustfs-manage.yml --tags config
+    ansible-playbook -f 4 -l bench rustfs-manage.yml --tags binary-copy
+    ansible-playbook -f 4 -l bench rustfs-manage.yml --tags start
+  ' \
+  --duration 180s \
+  --rounds 5 \
+  --cooldown 45 \
+  --concurrency 32 \
+  --out-dir target/hotpath-abba/cluster-pr-XXXX
+```
+
+For formal evidence, prefer `--rounds 5` or higher when the cluster budget
+allows it. The script enforces `--rounds >= 3`.
+
+## CPU and memory evidence
+
+ABBA warp output answers whether the candidate changed throughput or latency.
+Collect host telemetry at the same time to explain why.
+
+Recommended minimum:
+
+```bash
+mkdir -p target/hotpath-abba/cluster-pr-XXXX/telemetry
+
+pidstat -durh 5 > target/hotpath-abba/cluster-pr-XXXX/telemetry/pidstat.txt &
+PIDSTAT_PID=$!
+
+mpstat 5 > target/hotpath-abba/cluster-pr-XXXX/telemetry/mpstat.txt &
+MPSTAT_PID=$!
+
+iostat -xz 5 > target/hotpath-abba/cluster-pr-XXXX/telemetry/iostat.txt &
+IOSTAT_PID=$!
+```
+
+Stop the collectors after the ABBA script exits:
+
+```bash
+kill "$PIDSTAT_PID" "$MPSTAT_PID" "$IOSTAT_PID"
+```
+
+For deeper CPU attribution, run `perf record` around one representative
+workload after the ABBA gate identifies a candidate regression or improvement:
+
+```bash
+perf record -F 99 -g -- sleep 180
+perf report --stdio > target/hotpath-abba/cluster-pr-XXXX/telemetry/perf-report.txt
+```
+
+For allocation profiling, build the candidate with:
+
+```bash
+cargo build --release -p rustfs --bins --features hotpath-alloc
+```
+
+Then run the same ABBA command with that binary. Compare allocation-heavy
+function sections only within the same build mode. Do not compare
+`hotpath-alloc` binaries directly with default release binaries for throughput
+acceptance, because allocation instrumentation intentionally changes what is
+measured.
+
+For CPU hotpath sections emitted by hotpath, build with:
+
+```bash
+cargo build --release -p rustfs --bins --features hotpath-cpu
+```
+
+Use the CPU-enabled report to explain hotspots after the default or plain
+`hotpath` ABBA gate shows a real effect.
+
+## Output layout
+
+The ABBA runner writes:
+
+```text
+<out-dir>/
+  manifest.env
+  abba_schedule.csv
+  candidate_gate.md
+  baseline_drift_gate.md
+  summary.md
+  <workload>/<sync>/<leg>/median_summary.csv
+  <workload>/<sync>/<leg>/baseline_compare.csv
+```
+
+Attach or link at least these files in the issue or PR:
+
+- `summary.md`
+- `candidate_gate.md`
+- `baseline_drift_gate.md`
+- `abba_schedule.csv`
+- every `median_summary.csv` and `baseline_compare.csv` for a failed or
+  borderline workload
+- host telemetry files used to explain CPU, memory, or disk saturation
+
+## Interpretation
+
+Use this decision table:
+
+| Candidate gate | A2 drift gate | Interpretation |
+| --- | --- | --- |
+| PASS | PASS | Candidate is acceptable for the measured matrix. |
+| WARN | PASS | Candidate has a small measurable signal; inspect telemetry and decide if it is expected. |
+| FAIL | PASS | Candidate likely regressed the affected workload; investigate before merge. |
+| FAIL | FAIL on the same workload | Environment drift is high; rerun on a quieter runner or increase duration and rounds. |
+| PASS | FAIL | Candidate did not exceed the budget, but the rig was unstable; avoid using the numbers as proof of improvement. |
+
+When `B1` and `B2` disagree, treat the result as inconclusive even if the gate
+passes. Increase duration, rounds, cooldown, or runner isolation before drawing
+a conclusion.
+
+## AI execution checklist
+
+When delegating the run to an AI agent or an automation runner, provide these
+inputs explicitly:
+
+- repository checkout and candidate branch or commit;
+- baseline commit or binary path;
+- candidate binary path;
+- runner type: local Linux or external cluster;
+- endpoint, access key, secret key source, and region;
+- deploy hook path or exact command for cluster mode;
+- output directory;
+- required duration, rounds, cooldown, concurrency, and fail/warn budgets;
+- where to upload artifacts after the run.
+
+The AI agent should execute this sequence:
+
+1. Confirm `uname -a`, RustFS commits, binary SHA256 sums, `warp --version`,
+   CPU model, memory size, disk layout, and whether the run is local or cluster.
+2. Run `scripts/run_hotpath_warp_abba.sh --dry-run` with the final arguments.
+3. Run the real ABBA command with `--rounds >= 3`.
+4. Preserve the full output directory without editing generated CSV files.
+5. Read `summary.md`, `candidate_gate.md`, and `baseline_drift_gate.md`.
+6. Summarize only measured facts: candidate deltas, baseline drift, CPU or
+   memory saturation, and any failed workloads.
+7. Post the summary and artifact location to the tracking issue or PR.
+
+Do not report a performance win or loss when the baseline drift gate failed on
+the same workload and no rerun was collected.

rustfs/Cargo.toml

@@ -60,25 +60,137 @@ hotpath = [
     "hotpath/crossbeam",
     "hotpath/parking_lot",
     "hotpath/reqwest-0-13",
+    "rustfs-audit/hotpath",
+    "rustfs-common/hotpath",
+    "rustfs-concurrency/hotpath",
+    "rustfs-config/hotpath",
+    "rustfs-credentials/hotpath",
+    "rustfs-crypto/hotpath",
+    "rustfs-data-usage/hotpath",
     "rustfs-ecstore/hotpath",
+    "rustfs-extension-schema/hotpath",
     "rustfs-filemeta/hotpath",
+    "rustfs-heal/hotpath",
+    "rustfs-iam/hotpath",
+    "rustfs-io-core/hotpath",
+    "rustfs-io-metrics/hotpath",
+    "rustfs-keystone/hotpath",
+    "rustfs-kms/hotpath",
+    "rustfs-lock/hotpath",
+    "rustfs-log-analyzer/hotpath",
+    "rustfs-madmin/hotpath",
+    "rustfs-notify/hotpath",
+    "rustfs-object-capacity/hotpath",
+    "rustfs-object-data-cache/hotpath",
+    "rustfs-obs/hotpath",
     "rustfs-policy/hotpath",
+    "rustfs-protocols/hotpath",
+    "rustfs-protos/hotpath",
     "rustfs-rio/hotpath",
+    "rustfs-s3-ops/hotpath",
+    "rustfs-s3-types/hotpath",
+    "rustfs-s3select-api/hotpath",
+    "rustfs-s3select-query/hotpath",
+    "rustfs-scanner/hotpath",
+    "rustfs-security-governance/hotpath",
+    "rustfs-signer/hotpath",
+    "rustfs-storage-api/hotpath",
+    "rustfs-targets/hotpath",
+    "rustfs-tls-runtime/hotpath",
+    "rustfs-trusted-proxies/hotpath",
+    "rustfs-utils/hotpath",
+    "rustfs-zip/hotpath",
+    "rustfs-test-utils/hotpath",
 ]
 hotpath-alloc = [
+    "hotpath",
     "hotpath/hotpath-alloc",
+    "rustfs-audit/hotpath-alloc",
+    "rustfs-common/hotpath-alloc",
+    "rustfs-concurrency/hotpath-alloc",
+    "rustfs-config/hotpath-alloc",
+    "rustfs-credentials/hotpath-alloc",
+    "rustfs-crypto/hotpath-alloc",
+    "rustfs-data-usage/hotpath-alloc",
     "rustfs-ecstore/hotpath-alloc",
+    "rustfs-extension-schema/hotpath-alloc",
     "rustfs-filemeta/hotpath-alloc",
+    "rustfs-heal/hotpath-alloc",
+    "rustfs-iam/hotpath-alloc",
+    "rustfs-io-core/hotpath-alloc",
+    "rustfs-io-metrics/hotpath-alloc",
+    "rustfs-keystone/hotpath-alloc",
+    "rustfs-kms/hotpath-alloc",
+    "rustfs-lock/hotpath-alloc",
+    "rustfs-log-analyzer/hotpath-alloc",
+    "rustfs-madmin/hotpath-alloc",
+    "rustfs-notify/hotpath-alloc",
+    "rustfs-object-capacity/hotpath-alloc",
+    "rustfs-object-data-cache/hotpath-alloc",
+    "rustfs-obs/hotpath-alloc",
     "rustfs-policy/hotpath-alloc",
+    "rustfs-protocols/hotpath-alloc",
+    "rustfs-protos/hotpath-alloc",
     "rustfs-rio/hotpath-alloc",
+    "rustfs-s3-ops/hotpath-alloc",
+    "rustfs-s3-types/hotpath-alloc",
+    "rustfs-s3select-api/hotpath-alloc",
+    "rustfs-s3select-query/hotpath-alloc",
+    "rustfs-scanner/hotpath-alloc",
+    "rustfs-security-governance/hotpath-alloc",
+    "rustfs-signer/hotpath-alloc",
+    "rustfs-storage-api/hotpath-alloc",
+    "rustfs-targets/hotpath-alloc",
+    "rustfs-tls-runtime/hotpath-alloc",
+    "rustfs-trusted-proxies/hotpath-alloc",
+    "rustfs-utils/hotpath-alloc",
+    "rustfs-zip/hotpath-alloc",
+    "rustfs-test-utils/hotpath-alloc",
 ]
 hotpath-cpu = [
     "hotpath",
     "hotpath/hotpath-cpu",
+    "rustfs-audit/hotpath-cpu",
+    "rustfs-common/hotpath-cpu",
+    "rustfs-concurrency/hotpath-cpu",
+    "rustfs-config/hotpath-cpu",
+    "rustfs-credentials/hotpath-cpu",
+    "rustfs-crypto/hotpath-cpu",
+    "rustfs-data-usage/hotpath-cpu",
     "rustfs-ecstore/hotpath-cpu",
+    "rustfs-extension-schema/hotpath-cpu",
     "rustfs-filemeta/hotpath-cpu",
+    "rustfs-heal/hotpath-cpu",
+    "rustfs-iam/hotpath-cpu",
+    "rustfs-io-core/hotpath-cpu",
+    "rustfs-io-metrics/hotpath-cpu",
+    "rustfs-keystone/hotpath-cpu",
+    "rustfs-kms/hotpath-cpu",
+    "rustfs-lock/hotpath-cpu",
+    "rustfs-log-analyzer/hotpath-cpu",
+    "rustfs-madmin/hotpath-cpu",
+    "rustfs-notify/hotpath-cpu",
+    "rustfs-object-capacity/hotpath-cpu",
+    "rustfs-object-data-cache/hotpath-cpu",
+    "rustfs-obs/hotpath-cpu",
     "rustfs-policy/hotpath-cpu",
+    "rustfs-protocols/hotpath-cpu",
+    "rustfs-protos/hotpath-cpu",
     "rustfs-rio/hotpath-cpu",
+    "rustfs-s3-ops/hotpath-cpu",
+    "rustfs-s3-types/hotpath-cpu",
+    "rustfs-s3select-api/hotpath-cpu",
+    "rustfs-s3select-query/hotpath-cpu",
+    "rustfs-scanner/hotpath-cpu",
+    "rustfs-security-governance/hotpath-cpu",
+    "rustfs-signer/hotpath-cpu",
+    "rustfs-storage-api/hotpath-cpu",
+    "rustfs-targets/hotpath-cpu",
+    "rustfs-tls-runtime/hotpath-cpu",
+    "rustfs-trusted-proxies/hotpath-cpu",
+    "rustfs-utils/hotpath-cpu",
+    "rustfs-zip/hotpath-cpu",
+    "rustfs-test-utils/hotpath-cpu",
 ]
 
 [lints]

rustfs/src/admin/router.rs

@@ -2708,18 +2708,16 @@ impl<T: Operation> S3Router<T> {
 
     pub fn insert(&mut self, method: Method, path: &str, operation: T) -> std::io::Result<()> {
         let path = Self::make_route_str(method, path);
+        #[cfg(test)]
+        let registered_path = path.clone();
 
         // warn!("set uri {}", &path);
 
-        #[cfg(test)]
-        {
-            self.router.insert(path.clone(), operation).map_err(std::io::Error::other)?;
-            self.registered_routes.push(path);
-        }
-
-        #[cfg(not(test))]
         self.router.insert(path, operation).map_err(std::io::Error::other)?;
 
+        #[cfg(test)]
+        self.registered_routes.push(registered_path);
+
         Ok(())
     }
 

scripts/run_hotpath_warp_abba.sh

@@ -0,0 +1,428 @@
+#!/usr/bin/env bash
+# Formal Linux / production-cluster ABBA runner for the hotpath warp matrix.
+#
+# This script is intentionally a thin orchestrator around the existing
+# run_object_batch_bench_enhanced.sh load driver and hotpath_warp_ab_gate.sh
+# relative-budget gate. It runs each durability/workload cell as:
+#
+#   A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline
+#
+# Candidate legs are compared against A1. The final A2 leg is also compared
+# against A1 to quantify baseline drift separately from candidate deltas.
+
+set -euo pipefail
+
+PROJECT_ROOT="$(git rev-parse --show-toplevel)"
+ENHANCED_BENCH="${PROJECT_ROOT}/scripts/run_object_batch_bench_enhanced.sh"
+GATE="${PROJECT_ROOT}/scripts/hotpath_warp_ab_gate.sh"
+
+BASELINE_BIN=""
+CANDIDATE_BIN=""
+ENDPOINT=""
+DEPLOY_HOOK=""
+HEALTH_PATH="/health"
+ADDRESS="127.0.0.1:9000"
+DATA_ROOT="/tmp/rustfs-hotpath-abba"
+DISKS=4
+ACCESS_KEY="rustfsadmin"
+SECRET_KEY="rustfsadmin"
+REGION="us-east-1"
+WARP_BIN="warp"
+CONCURRENCY=8
+DURATION="60s"
+ROUNDS=3
+COOLDOWN_SECS=20
+HEALTH_TIMEOUT_SECS=180
+FAIL_PCT=10
+WARN_PCT=5
+ALLOW_REGRESSION=false
+EXEMPTION_REASON="deliberate correctness tradeoff"
+OUT_DIR="${PROJECT_ROOT}/target/hotpath-abba/$(date -u +%Y%m%dT%H%M%SZ 2>/dev/null || echo run)"
+DRY_RUN=false
+
+WORKLOADS=(
+  "put-4kib|put|4KiB"
+  "put-4mib|put|4MiB"
+  "get-4kib|get|4KiB"
+  "get-4mib|get|4MiB"
+  "get-10mib|get|10MiB"
+  "mixed-256k|mixed|256KiB"
+)
+DRIVE_SYNC_MATRIX=("sync-on|true" "sync-off|false")
+
+usage() {
+  cat <<'USAGE'
+Usage: scripts/run_hotpath_warp_abba.sh --baseline-bin <path> --candidate-bin <path> [options]
+
+Formal ABBA mode for Linux runners or production-like clusters. The schedule is
+A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline for every workload
+and drive-sync cell.
+
+Required:
+  --baseline-bin <path>     Baseline RustFS binary.
+  --candidate-bin <path>    Candidate RustFS binary.
+
+Local Linux runner mode:
+  --address <host:port>     Local RustFS address (default 127.0.0.1:9000).
+  --disks <n>               Throwaway local disks per node (default 4).
+  --data-root <path>        Local disk root (default /tmp/rustfs-hotpath-abba).
+
+Production / cluster mode:
+  --endpoint <host:port>    Existing cluster endpoint. Enables external mode.
+  --deploy-hook <cmd>       Command run before each ABBA leg. It receives:
+                              HOTPATH_ABBA_LEG=A1|B1|B2|A2
+                              HOTPATH_ABBA_PHASE=baseline|candidate
+                              HOTPATH_ABBA_BINARY=<baseline/candidate binary>
+                              HOTPATH_ABBA_DRIVE_SYNC=true|false
+  --health-path <path>      Readiness path (default /health).
+
+Benchmark:
+  --duration <dur>          warp duration per cell (default 60s).
+  --rounds <n>              rounds per cell; must be >= 3 (default 3).
+  --cooldown <n>            cooldown seconds between rounds/sizes (default 20).
+  --concurrency <n>         warp concurrency (default 8).
+  --warp-bin <path>         warp binary (default warp).
+
+Credentials:
+  --access-key <value>      S3 access key (default rustfsadmin).
+  --secret-key <value>      S3 secret key (default rustfsadmin).
+  --region <value>          S3 region (default us-east-1).
+
+Gate:
+  --fail-pct <n>            Regression budget that fails gate (default 10).
+  --warn-pct <n>            Regression budget that warns (default 5).
+  --allow-regression        Downgrade candidate gate FAIL to WARN.
+  --exemption-reason <s>    Reason recorded when allow-regression is used.
+
+Output:
+  --out-dir <path>          Output dir (default target/hotpath-abba/<ts>).
+  --dry-run                 Print commands without starting servers or warp.
+  -h, --help
+
+Outputs:
+  <out-dir>/abba_schedule.csv
+  <out-dir>/candidate_gate.md
+  <out-dir>/baseline_drift_gate.md
+  <out-dir>/summary.md
+  <out-dir>/<workload>/<sync>/<leg>/{median_summary.csv,baseline_compare.csv}
+USAGE
+}
+
+die() {
+  echo "error: $*" >&2
+  exit 2
+}
+
+log() {
+  printf '[hotpath-abba] %s\n' "$*" >&2
+}
+
+run() {
+  if [[ "$DRY_RUN" == "true" ]]; then
+    { printf 'DRY-RUN:'; printf ' %q' "$@"; printf '\n'; } >&2
+    return 0
+  fi
+  "$@"
+}
+
+validate_positive_int() {
+  local value="$1" name="$2"
+  [[ "$value" =~ ^[0-9]+$ && "$value" -gt 0 ]] || die "$name must be a positive integer"
+}
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --baseline-bin) BASELINE_BIN="$2"; shift 2 ;;
+    --candidate-bin) CANDIDATE_BIN="$2"; shift 2 ;;
+    --endpoint) ENDPOINT="$2"; shift 2 ;;
+    --deploy-hook) DEPLOY_HOOK="$2"; shift 2 ;;
+    --health-path) HEALTH_PATH="$2"; shift 2 ;;
+    --address) ADDRESS="$2"; shift 2 ;;
+    --data-root) DATA_ROOT="$2"; shift 2 ;;
+    --disks) DISKS="$2"; shift 2 ;;
+    --access-key) ACCESS_KEY="$2"; shift 2 ;;
+    --secret-key) SECRET_KEY="$2"; shift 2 ;;
+    --region) REGION="$2"; shift 2 ;;
+    --warp-bin) WARP_BIN="$2"; shift 2 ;;
+    --concurrency) CONCURRENCY="$2"; shift 2 ;;
+    --duration) DURATION="$2"; shift 2 ;;
+    --rounds) ROUNDS="$2"; shift 2 ;;
+    --cooldown) COOLDOWN_SECS="$2"; shift 2 ;;
+    --health-timeout) HEALTH_TIMEOUT_SECS="$2"; shift 2 ;;
+    --fail-pct) FAIL_PCT="$2"; shift 2 ;;
+    --warn-pct) WARN_PCT="$2"; shift 2 ;;
+    --allow-regression) ALLOW_REGRESSION=true; shift ;;
+    --exemption-reason) EXEMPTION_REASON="$2"; shift 2 ;;
+    --out-dir) OUT_DIR="$2"; shift 2 ;;
+    --dry-run) DRY_RUN=true; shift ;;
+    -h|--help) usage; exit 0 ;;
+    *) die "unknown argument: $1" ;;
+  esac
+done
+
+validate_positive_int "$DISKS" "--disks"
+validate_positive_int "$CONCURRENCY" "--concurrency"
+validate_positive_int "$ROUNDS" "--rounds"
+validate_positive_int "$COOLDOWN_SECS" "--cooldown"
+validate_positive_int "$HEALTH_TIMEOUT_SECS" "--health-timeout"
+validate_positive_int "$FAIL_PCT" "--fail-pct"
+validate_positive_int "$WARN_PCT" "--warn-pct"
+[[ "$ROUNDS" -ge 3 ]] || die "--rounds must be >= 3 for formal ABBA evidence"
+[[ -n "$BASELINE_BIN" ]] || die "--baseline-bin is required"
+[[ -n "$CANDIDATE_BIN" ]] || die "--candidate-bin is required"
+[[ "$DRY_RUN" == "true" || -x "$BASELINE_BIN" ]] || die "baseline binary is not executable: $BASELINE_BIN"
+[[ "$DRY_RUN" == "true" || -x "$CANDIDATE_BIN" ]] || die "candidate binary is not executable: $CANDIDATE_BIN"
+[[ -x "$ENHANCED_BENCH" ]] || die "missing load driver: $ENHANCED_BENCH"
+[[ -x "$GATE" ]] || die "missing gate: $GATE"
+if [[ "$DRY_RUN" != "true" ]] && ! command -v "$WARP_BIN" >/dev/null 2>&1; then
+  die "warp not found on PATH; install warp or pass --warp-bin"
+fi
+
+EXTERNAL=false
+if [[ -n "$ENDPOINT" ]]; then
+  EXTERNAL=true
+  ADDRESS="$ENDPOINT"
+  [[ -n "$DEPLOY_HOOK" ]] || log "warning: external mode without --deploy-hook; binaries must be swapped out of band"
+fi
+
+mkdir -p "$OUT_DIR"
+SERVER_LOG_DIR="$OUT_DIR/server-logs"
+run mkdir -p "$SERVER_LOG_DIR"
+
+SERVER_PID=""
+SERVER_LOG=""
+tear_down() {
+  [[ "$EXTERNAL" == "true" ]] && return 0
+  [[ -n "$SERVER_PID" ]] || return 0
+  run kill "$SERVER_PID" 2>/dev/null || true
+  SERVER_PID=""
+}
+trap tear_down EXIT INT TERM
+
+dump_server_log() {
+  [[ -n "$SERVER_LOG" && -f "$SERVER_LOG" ]] || return 0
+  echo "----- last 80 lines of $SERVER_LOG -----" >&2
+  tail -n 80 "$SERVER_LOG" >&2 || true
+  echo "----------------------------------------" >&2
+}
+
+wait_health() {
+  [[ "$DRY_RUN" == "true" ]] && return 0
+  local i
+  for ((i = 0; i < HEALTH_TIMEOUT_SECS; i++)); do
+    if [[ "$EXTERNAL" != "true" && -n "$SERVER_PID" ]] && ! kill -0 "$SERVER_PID" 2>/dev/null; then
+      echo "error: rustfs server (pid $SERVER_PID) exited before becoming healthy after ${i}s" >&2
+      dump_server_log
+      return 1
+    fi
+    if curl -fsS "http://${ADDRESS}${HEALTH_PATH}" >/dev/null 2>&1; then
+      return 0
+    fi
+    sleep 1
+  done
+  echo "error: endpoint http://${ADDRESS}${HEALTH_PATH} did not become healthy within ${HEALTH_TIMEOUT_SECS}s" >&2
+  dump_server_log
+  return 1
+}
+
+binary_for_leg() {
+  case "$1" in
+    A1|A2) echo "$BASELINE_BIN" ;;
+    B1|B2) echo "$CANDIDATE_BIN" ;;
+    *) die "unknown ABBA leg: $1" ;;
+  esac
+}
+
+phase_for_leg() {
+  case "$1" in
+    A1|A2) echo "baseline" ;;
+    B1|B2) echo "candidate" ;;
+    *) die "unknown ABBA leg: $1" ;;
+  esac
+}
+
+bring_up() {
+  local leg="$1" drive_sync="$2"
+  local phase bin
+  phase="$(phase_for_leg "$leg")"
+  bin="$(binary_for_leg "$leg")"
+
+  if [[ "$EXTERNAL" == "true" ]]; then
+    if [[ -n "$DEPLOY_HOOK" ]]; then
+      log "deploy hook: leg=$leg phase=$phase drive_sync=$drive_sync"
+      HOTPATH_ABBA_LEG="$leg" HOTPATH_ABBA_PHASE="$phase" HOTPATH_ABBA_BINARY="$bin" HOTPATH_ABBA_DRIVE_SYNC="$drive_sync" \
+        run bash -c "$DEPLOY_HOOK"
+    fi
+    wait_health
+    return 0
+  fi
+
+  local node_dir="$DATA_ROOT/$leg-sync-$drive_sync"
+  local disks=() d
+  for ((d = 1; d <= DISKS; d++)); do
+    disks+=("$node_dir/d$d")
+  done
+  run mkdir -p "${disks[@]}"
+  SERVER_LOG="$SERVER_LOG_DIR/$leg-sync-$drive_sync.log"
+
+  if [[ "$DRY_RUN" == "true" ]]; then
+    { printf 'DRY-RUN: RUSTFS_DRIVE_SYNC_ENABLE=%s %q server' "$drive_sync" "$bin"
+      printf ' %q' "${disks[@]}"; printf '\n'; } >&2
+    SERVER_PID="dry-run"
+    return 0
+  fi
+
+  cat >"$SERVER_LOG_DIR/$leg-sync-$drive_sync.env" <<EOF
+leg=$leg
+phase=$phase
+drive_sync=$drive_sync
+binary=$bin
+address=$ADDRESS
+disks=${disks[*]}
+health_url=http://${ADDRESS}${HEALTH_PATH}
+health_timeout_secs=$HEALTH_TIMEOUT_SECS
+uname=$(uname -a 2>/dev/null || echo unknown)
+warp_version=$("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
+EOF
+  RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true \
+  RUSTFS_ADDRESS="$ADDRESS" \
+  RUSTFS_ACCESS_KEY="$ACCESS_KEY" \
+  RUSTFS_SECRET_KEY="$SECRET_KEY" \
+  RUSTFS_REGION="$REGION" \
+  RUSTFS_CONSOLE_ENABLE=false \
+  RUSTFS_DRIVE_SYNC_ENABLE="$drive_sync" \
+    "$bin" server "${disks[@]}" >"$SERVER_LOG" 2>&1 &
+  SERVER_PID=$!
+  wait_health
+}
+
+measure() {
+  local leg="$1" workload="$2" mode="$3" size="$4" sync_label="$5" baseline_csv="${6:-}"
+  local cell="$OUT_DIR/$workload/$sync_label/$leg"
+  local args=(
+    --tool warp --warp-bin "$WARP_BIN" --warp-mode "$mode"
+    --endpoint "$ADDRESS" --access-key "$ACCESS_KEY" --secret-key "$SECRET_KEY"
+    --region "$REGION" --sizes "$size" --concurrency "$CONCURRENCY"
+    --duration "$DURATION" --rounds "$ROUNDS" --cooldown-secs "$COOLDOWN_SECS"
+    --out-dir "$cell"
+  )
+  [[ -n "$baseline_csv" ]] && args+=(--baseline-csv "$baseline_csv")
+  run "$ENHANCED_BENCH" "${args[@]}" >&2
+  echo "$cell"
+}
+
+write_schedule_header() {
+  echo "sync_label,drive_sync,workload,mode,size,leg,phase,binary,out_dir" >"$OUT_DIR/abba_schedule.csv"
+}
+
+append_schedule() {
+  local sync_label="$1" drive_sync="$2" workload="$3" mode="$4" size="$5" leg="$6"
+  local phase bin
+  phase="$(phase_for_leg "$leg")"
+  bin="$(binary_for_leg "$leg")"
+  echo "$sync_label,$drive_sync,$workload,$mode,$size,$leg,$phase,$bin,$OUT_DIR/$workload/$sync_label/$leg" >>"$OUT_DIR/abba_schedule.csv"
+}
+
+write_manifest() {
+  cat >"$OUT_DIR/manifest.env" <<EOF
+generated_at_utc=$(date -u +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || echo unknown)
+runner=$(uname -srm 2>/dev/null || echo unknown)
+schedule=ABBA
+rounds=$ROUNDS
+duration=$DURATION
+cooldown_secs=$COOLDOWN_SECS
+concurrency=$CONCURRENCY
+baseline_bin=$BASELINE_BIN
+candidate_bin=$CANDIDATE_BIN
+external=$EXTERNAL
+endpoint=$ADDRESS
+warp_version=$("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
+EOF
+}
+
+declare -a CANDIDATE_COMPARE_CSVS=()
+declare -a DRIFT_COMPARE_CSVS=()
+
+write_manifest
+write_schedule_header
+
+for ds_spec in "${DRIVE_SYNC_MATRIX[@]}"; do
+  IFS='|' read -r sync_label drive_sync <<<"$ds_spec"
+
+  for leg in A1 B1 B2 A2; do
+    log "=== $sync_label leg $leg ($(phase_for_leg "$leg")) ==="
+    bring_up "$leg" "$drive_sync"
+
+    for wl_spec in "${WORKLOADS[@]}"; do
+      IFS='|' read -r workload mode size <<<"$wl_spec"
+      append_schedule "$sync_label" "$drive_sync" "$workload" "$mode" "$size" "$leg"
+
+      baseline_csv=""
+      if [[ "$leg" != "A1" ]]; then
+        baseline_csv="$OUT_DIR/$workload/$sync_label/A1/median_summary.csv"
+      fi
+
+      cell="$(measure "$leg" "$workload" "$mode" "$size" "$sync_label" "$baseline_csv")"
+      case "$leg" in
+        B1|B2) CANDIDATE_COMPARE_CSVS+=("$cell/baseline_compare.csv") ;;
+        A2) DRIFT_COMPARE_CSVS+=("$cell/baseline_compare.csv") ;;
+      esac
+    done
+
+    tear_down
+  done
+done
+
+gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --markdown "$OUT_DIR/candidate_gate.md")
+for csv in "${CANDIDATE_COMPARE_CSVS[@]}"; do
+  gate_args+=(--compare-csv "$csv")
+done
+[[ "$ALLOW_REGRESSION" == "true" ]] && gate_args+=(--allow-regression --exemption-reason "$EXEMPTION_REASON")
+
+drift_gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --markdown "$OUT_DIR/baseline_drift_gate.md")
+for csv in "${DRIFT_COMPARE_CSVS[@]}"; do
+  drift_gate_args+=(--compare-csv "$csv")
+done
+
+if [[ "$DRY_RUN" == "true" ]]; then
+  log "dry-run complete; candidate compare CSVs=${#CANDIDATE_COMPARE_CSVS[@]} baseline drift CSVs=${#DRIFT_COMPARE_CSVS[@]}"
+  { printf 'DRY-RUN:'; printf ' %q' "$GATE" "${gate_args[@]}"; printf '\n'; } >&2
+  { printf 'DRY-RUN:'; printf ' %q' "$GATE" "${drift_gate_args[@]}"; printf '\n'; } >&2
+  exit 0
+fi
+
+log "applying candidate relative-budget gate"
+set +e
+"$GATE" "${gate_args[@]}"
+candidate_status=$?
+set -e
+
+log "applying A2-vs-A1 baseline drift gate"
+set +e
+"$GATE" "${drift_gate_args[@]}"
+drift_status=$?
+set -e
+
+cat >"$OUT_DIR/summary.md" <<EOF
+# Hotpath Warp ABBA Summary
+
+- schedule: A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline
+- runner: $(uname -srm 2>/dev/null || echo unknown)
+- warp: $("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
+- matrix: duration=$DURATION rounds=$ROUNDS cooldown=$COOLDOWN_SECS disks=$DISKS concurrency=$CONCURRENCY
+- endpoint: $ADDRESS
+- baseline binary: $BASELINE_BIN
+- candidate binary: $CANDIDATE_BIN
+- candidate gate: $OUT_DIR/candidate_gate.md (exit $candidate_status)
+- baseline drift gate: $OUT_DIR/baseline_drift_gate.md (exit $drift_status)
+
+Interpretation:
+
+- Treat candidate gate failures as actionable only when the A2-vs-A1 drift gate is PASS or the affected workload's A2 drift is materially smaller than the B1/B2 candidate delta.
+- If both candidate and baseline drift fail on the same workload, rerun with longer duration, more rounds, or a quieter runner before assigning causality.
+EOF
+
+log "summary written to $OUT_DIR/summary.md"
+if [[ "$candidate_status" -ne 0 || "$drift_status" -ne 0 ]]; then
+  exit 1
+fi