test(s3): cover delete marker list visibility

Cargo.lock

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

crates/audit/Cargo.toml

@@ -25,33 +25,7 @@ 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,14 +28,7 @@ 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,14 +27,7 @@ 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,14 +13,7 @@ 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,7 +25,6 @@ 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"] }
@@ -35,9 +34,6 @@ 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,14 +24,7 @@ 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,7 +29,6 @@ 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 }
@@ -50,9 +49,6 @@ 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,14 +27,7 @@ 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,58 +25,10 @@ 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,6 +56,21 @@ 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() {
@@ -94,6 +109,7 @@ 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]
@@ -118,6 +134,17 @@ 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()
@@ -145,6 +172,7 @@ 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,83 +40,19 @@ 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/lib.rs

@@ -12,8 +12,6 @@
 // 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/extension-schema/Cargo.toml

@@ -27,14 +27,7 @@ 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", "rustfs-utils/hotpath"]
-hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
-hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
+hotpath = ["hotpath/hotpath", "hotpath/tokio"]
+hotpath-alloc = ["hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 
 [dependencies]
 hotpath.workspace = true

crates/heal/Cargo.toml

@@ -29,48 +29,7 @@ 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,7 +159,6 @@ 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,7 +584,6 @@ 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();
@@ -760,7 +759,6 @@ 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",
@@ -1406,7 +1404,6 @@ 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,55 +28,7 @@ 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,7 +469,6 @@ impl ObjectStore {
         });
     }
 
-    #[hotpath::measure]
     async fn list_all_iamconfig_items(&self) -> Result<HashMap<String, Vec<String>>> {
         let (tx, mut rx) = mpsc::channel::<StringOrErr>(100);
 
@@ -509,7 +508,6 @@ 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,14 +27,7 @@ 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,32 +28,7 @@ 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,34 +28,7 @@ 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,7 +95,6 @@ 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,
@@ -239,7 +238,6 @@ 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,7 +28,6 @@ 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"] }
@@ -85,17 +84,3 @@ 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,24 +46,6 @@ 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!(
@@ -135,9 +117,11 @@ 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,
-/// entirely through the `KmsBackend` product surface.
-async fn assert_state_machine_contract(backend: &dyn KmsBackend, key_id: &str) {
+/// 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) {
     // Enabled: cryptographic use is allowed. Keep an envelope around to prove
     // decryption keeps working in later states.
     let data_key = backend
@@ -150,13 +134,16 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, key_id: &str) {
         .expect("Enabled key must encrypt");
 
     // Enabled -> Disabled.
-    backend.disable_key(key_id).await.expect("disable from Enabled must succeed");
+    client
+        .disable_key(key_id, None)
+        .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_rotate_rejected(backend, key_id).await;
+    expect_invalid_key_state(client.rotate_key(key_id, None).await, "");
     // ...but decryption of existing data keeps working (explicit AWS deviation)...
     let decrypted = backend
         .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
@@ -164,14 +151,17 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, key_id: &str) {
         .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.
-    backend.disable_key(key_id).await.expect("disable must be idempotent");
+    client.disable_key(key_id, None).await.expect("disable must be idempotent");
     expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "not pending deletion");
-    backend.enable_key(key_id).await.expect("enable from Disabled must succeed");
+    client
+        .enable_key(key_id, None)
+        .await
+        .expect("enable from Disabled must succeed");
     assert_key_state(backend, key_id, KeyState::Enabled).await;
 
     // Disabled keys may still be scheduled for deletion.
-    backend
-        .disable_key(key_id)
+    client
+        .disable_key(key_id, None)
         .await
         .expect("disable before scheduling must succeed");
     backend
@@ -183,9 +173,10 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, key_id: &str) {
     // 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(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(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.delete_key(schedule_request(key_id)).await, "pending deletion");
     let decrypted = backend
         .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
@@ -224,7 +215,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, &key_id).await;
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &key_id).await;
 }
 
 /// SSE-shaped regression: disabling a key must not break decryption of data
@@ -265,7 +256,10 @@ 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))
@@ -281,11 +275,9 @@ 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");
-    // 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);
+    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");
 }
 
 fn vault_dev_config(constructor: fn(url::Url, String) -> KmsConfig) -> KmsConfig {
@@ -306,15 +298,7 @@ async fn vault_kv2_backend_state_machine_contract() {
         .await
         .expect("key should be created");
 
-    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");
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &created.key_id).await;
 
     // Cleanup: leave the key pending deletion so repeated runs stay tidy.
     let _ = backend.delete_key(schedule_request(&created.key_id)).await;
@@ -332,12 +316,13 @@ async fn vault_transit_backend_state_machine_contract() {
         .await
         .expect("key should be created");
 
-    assert_state_machine_contract(&backend, &created.key_id).await;
+    assert_state_machine_contract(&backend, backend.lifecycle_client(), &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
-        .rotate_key(&created.key_id)
+        .lifecycle_client()
+        .rotate_key(&created.key_id, None)
         .await
         .expect("rotation of an Enabled transit key must succeed");
 

crates/kms/src/backends/local.rs

@@ -14,7 +14,9 @@
 
 //! Local file-based KMS backend implementation
 
-use crate::backends::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_status_permits};
+use crate::backends::{
+    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_status_permits,
+};
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -397,20 +399,6 @@ 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
@@ -477,7 +465,6 @@ 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)
@@ -520,7 +507,6 @@ impl LocalKmsClient {
             legacy_master_cipher,
             dek_crypto: AesDekCrypto::new(),
             key_write_locks: Mutex::new(HashMap::new()),
-            export_fence: Arc::new(tokio::sync::RwLock::new(())),
         })
     }
 
@@ -531,40 +517,12 @@ 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) -> 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;
+    async fn lock_key_for_write(&self, key_id: &str) -> tokio::sync::OwnedMutexGuard<()> {
         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())
         };
-        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)
+        lock.lock_owned().await
     }
 
     /// Derive a 256-bit key from the master key string using a persistent Argon2id salt.
@@ -841,11 +799,6 @@ 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()));
@@ -984,12 +937,9 @@ impl LocalKmsClient {
     }
 }
 
-impl LocalKmsClient {
-    pub(crate) async fn generate_data_key(
-        &self,
-        request: &GenerateKeyRequest,
-        context: Option<&OperationContext>,
-    ) -> Result<DataKeyInfo> {
+#[async_trait]
+impl KmsClient for LocalKmsClient {
+    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?;
@@ -1030,7 +980,7 @@ impl LocalKmsClient {
         Ok(data_key)
     }
 
-    pub(crate) async fn encrypt(&self, request: &EncryptRequest, context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    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
@@ -1047,7 +997,7 @@ impl LocalKmsClient {
         })
     }
 
-    pub(crate) async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
         debug!("Decrypting data");
 
         // Parse the data key envelope from ciphertext
@@ -1081,14 +1031,7 @@ impl LocalKmsClient {
         Ok(plaintext)
     }
 
-    /// 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> {
+    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
@@ -1117,18 +1060,14 @@ impl LocalKmsClient {
         Ok(master_key)
     }
 
-    pub(crate) async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
+    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())
     }
 
-    pub(crate) async fn list_keys(
-        &self,
-        request: &ListKeysRequest,
-        _context: Option<&OperationContext>,
-    ) -> Result<ListKeysResponse> {
+    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
         debug!("Listing keys");
 
         let mut keys = Vec::new();
@@ -1172,7 +1111,7 @@ impl LocalKmsClient {
         })
     }
 
-    pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    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;
@@ -1190,7 +1129,7 @@ impl LocalKmsClient {
         Ok(())
     }
 
-    pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    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;
@@ -1207,9 +1146,7 @@ impl LocalKmsClient {
         Ok(())
     }
 
-    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
-    #[cfg(test)]
-    pub(crate) async fn schedule_key_deletion(
+    async fn schedule_key_deletion(
         &self,
         key_id: &str,
         pending_window_days: u32,
@@ -1233,9 +1170,7 @@ impl LocalKmsClient {
         Ok(())
     }
 
-    /// 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<()> {
+    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;
@@ -1255,9 +1190,7 @@ impl LocalKmsClient {
         Ok(())
     }
 
-    /// 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> {
+    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));
         }
@@ -1266,7 +1199,7 @@ impl LocalKmsClient {
         ))
     }
 
-    pub(crate) async fn health_check(&self) -> Result<()> {
+    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"));
@@ -1277,6 +1210,17 @@ impl 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,6 +19,7 @@ use crate::types::*;
 use async_trait::async_trait;
 use jiff::Zoned;
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 
 #[cfg(test)]
 mod contract_tests;
@@ -98,6 +99,136 @@ 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 {
@@ -196,6 +327,58 @@ 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

@@ -61,7 +61,7 @@ impl ScriptedResponse {
 pub(crate) struct ScriptedVault {
     /// Base address (`http://127.0.0.1:port`) to point a Vault client at.
     pub(crate) address: String,
-    requests: Arc<Mutex<Vec<(String, String)>>>,
+    requests: Arc<Mutex<Vec<String>>>,
 }
 
 impl ScriptedVault {
@@ -81,10 +81,13 @@ impl ScriptedVault {
                 let Ok((mut stream, _)) = listener.accept().await else {
                     return;
                 };
-                let Some(request) = read_request(&mut stream).await else {
+                let Some(request_line) = read_request(&mut stream).await else {
                     continue;
                 };
-                recorded.lock().expect("scripted vault request log poisoned").push(request);
+                recorded
+                    .lock()
+                    .expect("scripted vault request log poisoned")
+                    .push(request_line);
                 let response = responses
                     .next()
                     .unwrap_or_else(|| ScriptedResponse::error(599, "scripted vault: script exhausted"));
@@ -104,32 +107,14 @@ impl ScriptedVault {
 
     /// The `METHOD /path` lines of every request served so far, in order.
     pub(crate) fn requests(&self) -> Vec<String> {
-        self.requests
-            .lock()
-            .expect("scripted vault request log poisoned")
-            .iter()
-            .map(|(line, _)| line.clone())
-            .collect()
-    }
-
-    /// The request bodies, in the same order as [`Self::requests`]; empty for
-    /// bodyless requests. Lets tests assert what a write actually persisted
-    /// (record contents, check-and-set options), not just that a write happened.
-    pub(crate) fn request_bodies(&self) -> Vec<String> {
-        self.requests
-            .lock()
-            .expect("scripted vault request log poisoned")
-            .iter()
-            .map(|(_, body)| body.clone())
-            .collect()
+        self.requests.lock().expect("scripted vault request log poisoned").clone()
     }
 }
 
 /// Read one HTTP/1.1 request (head plus content-length body) and return its
-/// `METHOD /path` line together with the body. Draining the body before
-/// responding keeps the client from seeing a connection reset while it is
-/// still writing.
-async fn read_request(stream: &mut TcpStream) -> Option<(String, String)> {
+/// `METHOD /path` line. Draining the body before responding keeps the client
+/// from seeing a connection reset while it is still writing.
+async fn read_request(stream: &mut TcpStream) -> Option<String> {
     let mut buffer = Vec::new();
     let mut chunk = [0u8; 4096];
     let head_end = loop {
@@ -161,17 +146,14 @@ async fn read_request(stream: &mut TcpStream) -> Option<(String, String)> {
         })
         .next()
         .unwrap_or(0);
-    let mut body = buffer[head_end..].to_vec();
-    let mut remaining = content_length.saturating_sub(body.len());
+    let mut remaining = content_length.saturating_sub(buffer.len() - head_end);
     while remaining > 0 {
         let read = stream.read(&mut chunk).await.ok()?;
         if read == 0 {
             break;
         }
-        body.extend_from_slice(&chunk[..read]);
         remaining = remaining.saturating_sub(read);
     }
-    body.truncate(content_length);
 
-    Some((format!("{method} {path}"), String::from_utf8_lossy(&body).into_owned()))
+    Some(format!("{method} {path}"))
 }

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": true,
+  "rotate": false,
   "schedule_deletion": true,
-  "versioning": true
+  "versioning": false
 }

crates/kms/src/backends/static_kms.rs

@@ -21,7 +21,7 @@
 //!
 //! encrypted_data(plaintext_len+16) || nonce (12 bytes)
 
-use crate::backends::{BackendCapabilities, KmsBackend};
+use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient};
 use crate::config::{BackendConfig, KmsConfig};
 use crate::encryption::DataKeyEnvelope;
 use crate::error::{KmsError, Result};
@@ -98,10 +98,9 @@ impl StaticKmsBackend {
     }
 }
 
-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> {
+#[async_trait]
+impl KmsClient for StaticKmsBackend {
+    async fn generate_data_key(&self, request: &GenerateKeyRequest, _context: Option<&OperationContext>) -> Result<DataKeyInfo> {
         if request.master_key_id != self.key_id {
             return Err(KmsError::key_not_found(&request.master_key_id));
         }
@@ -152,8 +151,7 @@ impl StaticKmsBackend {
         ))
     }
 
-    /// Encrypt caller-provided plaintext into the standard KMS envelope.
-    pub(crate) fn encrypt_to_envelope(&self, request: &EncryptRequest) -> Result<EncryptResponse> {
+    async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
         if request.key_id != self.key_id {
             return Err(KmsError::key_not_found(&request.key_id));
         }
@@ -198,8 +196,7 @@ impl StaticKmsBackend {
         })
     }
 
-    /// Open a KMS envelope produced by this backend.
-    pub(crate) fn decrypt_envelope(&self, request: &DecryptRequest) -> Result<Vec<u8>> {
+    async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> 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 {
@@ -238,8 +235,14 @@ impl StaticKmsBackend {
         Ok(plaintext)
     }
 
-    /// Describe the single configured key.
-    pub(crate) fn configured_key_info(&self, key_id: &str) -> Result<KeyInfo> {
+    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> {
         if key_id != self.key_id {
             return Err(KmsError::key_not_found(key_id));
         }
@@ -260,8 +263,7 @@ impl StaticKmsBackend {
         })
     }
 
-    /// List the single configured key, honouring the pagination marker.
-    pub(crate) fn list_configured_key(&self, request: &ListKeysRequest) -> Result<ListKeysResponse> {
+    async fn list_keys(&self, request: &ListKeysRequest, _context: Option<&OperationContext>) -> Result<ListKeysResponse> {
         let key_info = KeyInfo {
             key_id: self.key_id.clone(),
             description: Some("Static single-key KMS backend".to_string()),
@@ -293,6 +295,57 @@ impl 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]
@@ -306,12 +359,12 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn encrypt(&self, request: EncryptRequest) -> Result<EncryptResponse> {
-        self.encrypt_to_envelope(&request)
+        <Self as KmsClient>::encrypt(self, &request, None).await
     }
 
     async fn decrypt(&self, request: DecryptRequest) -> Result<DecryptResponse> {
         let key_id = self.key_id.clone();
-        let plaintext = self.decrypt_envelope(&request)?;
+        let plaintext = <Self as KmsClient>::decrypt(self, &request, None).await?;
         Ok(DecryptResponse {
             plaintext,
             key_id,
@@ -327,7 +380,7 @@ impl KmsBackend for StaticKmsBackend {
             encryption_context: request.encryption_context,
             grant_tokens: Vec::new(),
         };
-        let data_key = self.generate_data_key_envelope(&gen_req)?;
+        let data_key = <Self as KmsClient>::generate_data_key(self, &gen_req, None).await?;
 
         let plaintext_key = data_key
             .plaintext
@@ -342,7 +395,7 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn describe_key(&self, request: DescribeKeyRequest) -> Result<DescribeKeyResponse> {
-        let key_info = self.configured_key_info(&request.key_id)?;
+        let key_info = <Self as KmsClient>::describe_key(self, &request.key_id, None).await?;
         let key_metadata = KeyMetadata {
             key_id: key_info.key_id.clone(),
             key_state: if key_info.status == KeyStatus::Active {
@@ -362,7 +415,7 @@ impl KmsBackend for StaticKmsBackend {
     }
 
     async fn list_keys(&self, request: ListKeysRequest) -> Result<ListKeysResponse> {
-        self.list_configured_key(&request)
+        <Self as KmsClient>::list_keys(self, &request, None).await
     }
 
     async fn delete_key(&self, request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
@@ -393,7 +446,7 @@ impl KmsBackend for StaticKmsBackend {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::backends::KmsBackend as KmsBackendTrait;
+    use crate::backends::{KmsBackend as KmsBackendTrait, KmsClient};
     use crate::config::{BackendConfig, KmsBackend, StaticConfig};
     use crate::encryption::is_data_key_envelope;
     use base64::Engine as _;
@@ -438,8 +491,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 = backend
-            .generate_data_key_envelope(&request)
+        let data_key = KmsClient::generate_data_key(&backend, &request, None)
+            .await
             .expect("Failed to generate data key");
 
         assert_eq!(data_key.key_id, key_id);
@@ -456,7 +509,9 @@ 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 = backend.decrypt_envelope(&decrypt_request).expect("Failed to decrypt");
+        let decrypted = KmsClient::decrypt(&backend, &decrypt_request, None)
+            .await
+            .expect("Failed to decrypt");
 
         assert_eq!(decrypted.as_slice(), data_key.plaintext.as_deref().expect("plaintext should exist"));
     }
@@ -468,8 +523,8 @@ mod tests {
             .with_context("bucket".to_string(), "source-bucket".to_string())
             .with_context("object".to_string(), "source-object".to_string());
 
-        let data_key = backend
-            .generate_data_key_envelope(&request)
+        let data_key = KmsClient::generate_data_key(&backend, &request, None)
+            .await
             .expect("generate static KMS data key");
 
         assert!(
@@ -513,8 +568,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 = backend
-            .generate_data_key_envelope(&request)
+        let generated = KmsClient::generate_data_key(&backend, &request, None)
+            .await
             .expect("generate context-bound data key");
         let mut envelope: DataKeyEnvelope = serde_json::from_slice(&generated.ciphertext).expect("parse static KMS envelope");
         envelope
@@ -523,8 +578,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 = backend
-            .decrypt_envelope(&decrypt_request)
+        let error = KmsClient::decrypt(&backend, &decrypt_request, None)
+            .await
             .expect_err("tampering with authenticated envelope context must fail");
 
         assert!(matches!(error, KmsError::CryptographicError { .. }));
@@ -535,7 +590,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 = backend.generate_data_key_envelope(&request);
+        let result = KmsClient::generate_data_key(&backend, &request, None).await;
         assert!(result.is_err());
         assert!(result.expect_err("should be Err").to_string().contains("wrong-key-id"));
     }
@@ -547,7 +602,7 @@ mod tests {
         // Ciphertext too short
         let short = vec![0u8; 10];
         let request = DecryptRequest::new(short);
-        let result = backend.decrypt_envelope(&request);
+        let result = KmsClient::decrypt(&backend, &request, None).await;
         assert!(result.is_err());
     }
 
@@ -557,7 +612,9 @@ mod tests {
 
         // Generate a valid ciphertext first
         let gen_request = GenerateKeyRequest::new(key_id, "AES_256".to_string());
-        let data_key = backend.generate_data_key_envelope(&gen_request).expect("generate");
+        let data_key = KmsClient::generate_data_key(&backend, &gen_request, None)
+            .await
+            .expect("generate");
 
         // Tamper with the ciphertext (flip a bit in the encrypted portion)
         let mut tampered = data_key.ciphertext.clone();
@@ -566,7 +623,7 @@ mod tests {
         }
 
         let request = DecryptRequest::new(tampered);
-        let result = backend.decrypt_envelope(&request);
+        let result = KmsClient::decrypt(&backend, &request, None).await;
         assert!(result.is_err());
     }
 
@@ -575,14 +632,7 @@ mod tests {
         let (backend, key_id, _key) = create_test_backend().await;
 
         // Creating the pre-configured key should return KeyAlreadyExists
-        let result = KmsBackendTrait::create_key(
-            &backend,
-            CreateKeyRequest {
-                key_name: Some(key_id.clone()),
-                ..Default::default()
-            },
-        )
-        .await;
+        let result = KmsClient::create_key(&backend, &key_id, "AES_256", None).await;
         assert!(result.is_err());
         assert!(result.expect_err("should be Err").to_string().contains("already exists"));
     }
@@ -592,14 +642,7 @@ mod tests {
         let (backend, _key_id, _key) = create_test_backend().await;
 
         // Creating any other key should return invalid operation (read-only)
-        let result = KmsBackendTrait::create_key(
-            &backend,
-            CreateKeyRequest {
-                key_name: Some("other-key".to_string()),
-                ..Default::default()
-            },
-        )
-        .await;
+        let result = KmsClient::create_key(&backend, "other-key", "AES_256", None).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"));
@@ -609,13 +652,15 @@ mod tests {
     async fn test_describe_key() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let key_info = backend.configured_key_info(&key_id).expect("describe_key should succeed");
+        let key_info = KmsClient::describe_key(&backend, &key_id, None)
+            .await
+            .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 = backend.configured_key_info("nonexistent");
+        let result = KmsClient::describe_key(&backend, "nonexistent", None).await;
         assert!(result.is_err());
     }
 
@@ -623,8 +668,8 @@ mod tests {
     async fn test_list_keys() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let response = backend
-            .list_configured_key(&ListKeysRequest::default())
+        let response = KmsClient::list_keys(&backend, &ListKeysRequest::default(), None)
+            .await
             .expect("list_keys should succeed");
         assert_eq!(response.keys.len(), 1);
         assert_eq!(response.keys[0].key_id, key_id);
@@ -632,45 +677,61 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn lifecycle_mutations_are_unsupported_at_the_product_surface() {
+    async fn test_disable_key_returns_error() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        // 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:?}");
-        }
+        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());
     }
 
     #[tokio::test]
     async fn test_delete_key_returns_error() {
         let (backend, key_id, _key) = create_test_backend().await;
 
-        let result = KmsBackendTrait::delete_key(
-            &backend,
-            DeleteKeyRequest {
-                key_id: key_id.clone(),
-                pending_window_in_days: Some(7),
-                force_immediate: None,
-            },
-        )
-        .await;
+        let result = KmsClient::schedule_key_deletion(&backend, &key_id, 7, 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;
 
-        KmsBackendTrait::health_check(&backend)
-            .await
-            .expect("health_check should succeed");
+        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));
     }
 
     #[tokio::test]
@@ -679,13 +740,17 @@ mod tests {
 
         let plaintext = b"Hello, static KMS world!";
         let enc_request = EncryptRequest::new(key_id.clone(), plaintext.to_vec());
-        let enc_response = backend.encrypt_to_envelope(&enc_request).expect("encrypt should succeed");
+        let enc_response = KmsClient::encrypt(&backend, &enc_request, None)
+            .await
+            .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 = backend.decrypt_envelope(&dec_request).expect("decrypt should succeed");
+        let decrypted = KmsClient::decrypt(&backend, &dec_request, None)
+            .await
+            .expect("decrypt should succeed");
 
         assert_eq!(decrypted, plaintext);
     }

crates/kms/src/backends/vault_transit.rs

@@ -18,7 +18,9 @@ use crate::backends::vault_credentials::{
     CredentialTaskHandle, VaultClientHandle, VaultConnectionSettings, VaultCredentialPolicy, VaultCredentialProvider,
     token_source_for,
 };
-use crate::backends::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits};
+use crate::backends::{
+    BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits,
+};
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -504,12 +506,9 @@ impl VaultTransitKmsClient {
     }
 }
 
-impl VaultTransitKmsClient {
-    pub(crate) async fn generate_data_key(
-        &self,
-        request: &GenerateKeyRequest,
-        _context: Option<&OperationContext>,
-    ) -> Result<DataKeyInfo> {
+#[async_trait]
+impl KmsClient for VaultTransitKmsClient {
+    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?;
 
@@ -541,7 +540,7 @@ impl VaultTransitKmsClient {
         ))
     }
 
-    pub(crate) async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
+    async fn encrypt(&self, request: &EncryptRequest, _context: Option<&OperationContext>) -> Result<EncryptResponse> {
         let metadata = self
             .ensure_key_state_allows(&request.key_id, StateGatedOperation::Encrypt)
             .await?;
@@ -557,7 +556,7 @@ impl VaultTransitKmsClient {
         })
     }
 
-    pub(crate) async fn decrypt(&self, request: &DecryptRequest, _context: Option<&OperationContext>) -> Result<Vec<u8>> {
+    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}")))?;
 
@@ -579,14 +578,7 @@ impl VaultTransitKmsClient {
             .await
     }
 
-    /// 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> {
+    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));
         }
@@ -653,17 +645,11 @@ impl VaultTransitKmsClient {
         })
     }
 
-    /// 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> {
+    async fn describe_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<KeyInfo> {
         self.key_info(key_id).await
     }
 
-    pub(crate) async fn list_keys(
-        &self,
-        request: &ListKeysRequest,
-        _context: Option<&OperationContext>,
-    ) -> Result<ListKeysResponse> {
+    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)?;
@@ -706,7 +692,7 @@ impl VaultTransitKmsClient {
         })
     }
 
-    pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    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?;
@@ -715,15 +701,13 @@ impl VaultTransitKmsClient {
         self.store_key_metadata(key_id, &metadata).await
     }
 
-    pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> {
+    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
     }
 
-    /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
-    #[cfg(test)]
-    pub(crate) async fn schedule_key_deletion(
+    async fn schedule_key_deletion(
         &self,
         key_id: &str,
         pending_window_days: u32,
@@ -737,7 +721,17 @@ impl VaultTransitKmsClient {
         self.store_key_metadata(key_id, &metadata).await
     }
 
-    pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+    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> {
         self.ensure_key_state_allows(key_id, StateGatedOperation::Rotate).await?;
 
         // Single attempt, never retried: replaying a rotate whose response was
@@ -774,7 +768,7 @@ impl VaultTransitKmsClient {
         })
     }
 
-    pub(crate) async fn health_check(&self) -> Result<()> {
+    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)
@@ -786,6 +780,11 @@ impl 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 {
@@ -793,6 +792,14 @@ 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()?;
 
@@ -993,18 +1000,6 @@ 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)
     }
@@ -1442,58 +1437,4 @@ 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

@@ -1,998 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! 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,10 +355,8 @@ 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. 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.
+/// KEK, and which restore responsibility applies. Field order is part of the
+/// canonical digest form and is frozen for this format version.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct BackupManifest {
@@ -417,14 +415,8 @@ 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 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.
+    /// (unknown fields, duplicate fields, missing fields), then semantic
+    /// validation including digest verification.
     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 {
@@ -437,9 +429,7 @@ 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_pre_digest()?;
-        Self::verify_digest_in_bytes(bytes, &manifest.manifest_digest)?;
-        manifest.validate_content()?;
+        manifest.validate()?;
         Ok(manifest)
     }
 
@@ -459,30 +449,23 @@ impl BackupManifest {
         Ok(self)
     }
 
-    /// Compute the digest over the canonical manifest form.
+    /// Compute the digest over the canonical manifest bytes.
     ///
-    /// 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).
+    /// 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.
     pub fn compute_digest(&self) -> Result<ContentDigest, BackupError> {
         let mut unsealed = self.clone();
         unsealed.manifest_digest = ContentDigest::placeholder(self.manifest_digest.algorithm);
-        let value = serde_json::to_value(&unsealed)
+        let canonical = serde_json::to_vec(&unsealed)
             .map_err(|error| BackupError::corrupted(format!("manifest canonicalization failed: {error}")))?;
-        Self::digest_of_canonical_value(value, self.manifest_digest.algorithm)
+        match self.manifest_digest.algorithm {
+            DigestAlgorithm::Sha256 => Ok(ContentDigest::sha256_of(&canonical)),
+        }
     }
 
-    /// 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.
+    /// Verify the sealed digest against the current manifest content.
     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"));
@@ -495,33 +478,6 @@ 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
@@ -530,21 +486,11 @@ impl BackupManifest {
             .ok_or_else(|| BackupError::missing_artifact(artifact_kind_name(kind)))
     }
 
-    /// Validate the full manifest contract against the in-memory content.
+    /// Validate the full manifest contract.
     ///
-    /// 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.
+    /// This is decode-side validation and also guards [`Self::encode`], so a
+    /// producer cannot publish a manifest a decoder would reject.
     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,
@@ -554,12 +500,7 @@ impl BackupManifest {
         if self.completeness != CompletenessState::Complete {
             return Err(BackupError::incomplete_bundle("completeness marker records an in-progress bundle"));
         }
-        Ok(())
-    }
-
-    /// Semantic validation of everything except version, completeness, and
-    /// digest integrity.
-    fn validate_content(&self) -> Result<(), BackupError> {
+        self.verify_digest()?;
         require_non_empty("backup_id", &self.backup_id)?;
         require_non_empty("rustfs_version", &self.rustfs_version)?;
         require_non_empty("deployment_identity", &self.deployment_identity)?;
@@ -704,29 +645,6 @@ 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::*;
@@ -818,7 +736,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 = "a8b104d61c358cd75cc9a7691d2f7d2d96f73c53653bef8940a49e96dbdf7775";
+    const FIXTURE_DIGEST_HEX: &str = "01accb3e2bc51e12d17d1efc52ad6ab9441c50bec52c3fb29e0a9f92b725cdaa";
 
     fn fixture() -> String {
         FIXTURE.replace("SEALED_DIGEST_HEX", FIXTURE_DIGEST_HEX)
@@ -1092,39 +1010,12 @@ mod tests {
         let with_discovery = fixture().replace("\"completeness\"", "\"capability_discovery\": [1], \"completeness\"");
         expect_corrupted(BackupManifest::decode(with_discovery.as_bytes()), "reserved");
 
-        // 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.
+        // Explicit null carries no data and is tolerated as absence; digest
+        // verification still passes because null slots are skipped on
+        // serialization.
         let with_null = fixture().replace("\"completeness\"", "\"key_versions\": null, \"completeness\"");
-        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());
+        let decoded = BackupManifest::decode(with_null.as_bytes()).expect("null reserved slot should decode");
+        assert_eq!(decoded.key_versions, None);
     }
 
     #[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 contracts and backup production for KMS state.
+//! Backup/restore contract types for KMS state.
 //!
-//! 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.
+//! 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.
 //!
 //! # Bundle model
 //!
@@ -50,7 +50,6 @@
 mod capability;
 mod dry_run;
 mod error;
-pub mod local_export;
 mod manifest;
 
 pub use capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
@@ -58,10 +57,6 @@ 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,6 +182,7 @@ 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,19 +66,16 @@ 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

@@ -33,11 +33,9 @@ use std::time::Duration;
 
 use metrics_util::MetricKind;
 use metrics_util::debugging::{DebugValue, DebuggingRecorder};
-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,
-};
+use rustfs_kms::backends::KmsClient;
+use rustfs_kms::backends::vault::VaultKmsClient;
+use rustfs_kms::{KmsConfig, KmsError, VaultAuthMethod, VaultConfig};
 
 const OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
 const ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
@@ -56,23 +54,14 @@ fn vault_config(address: &str, token: &str) -> VaultConfig {
     }
 }
 
-fn kms_config(vault_config: VaultConfig, attempt_timeout: Duration, retry_attempts: u32) -> KmsConfig {
+fn kms_config(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>,
@@ -129,10 +118,11 @@ fn connection_refused_is_retried_within_budget() {
 
     let snapshot = record_metrics(|| {
         Box::pin(async move {
-            let client = VaultKmsBackend::new(kms_config(vault_config(&address, "unused"), Duration::from_secs(2), 2))
+            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 = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-refused"))
+            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:?}");
@@ -176,10 +166,11 @@ fn stalled_connection_is_cut_off_by_the_attempt_timeout() {
                 }
             });
 
-            let client = VaultKmsBackend::new(kms_config(vault_config(&address, "unused"), Duration::from_millis(250), 1))
+            let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_millis(250), 1))
                 .await
                 .expect("client construction performs no network calls");
-            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-stalled"))
+            let error = client
+                .describe_key("fault-injection-stalled", None)
                 .await
                 .expect_err("a stalled request must be cut off by the attempt timeout");
             assert!(
@@ -210,10 +201,11 @@ 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 = VaultKmsBackend::new(kms_config(config, Duration::from_secs(5), 3))
+            let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
                 .await
                 .expect("client construction performs no network calls");
-            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-forbidden"))
+            let error = client
+                .describe_key("fault-injection-forbidden", None)
                 .await
                 .expect_err("an invalid token must be rejected");
             assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
@@ -243,10 +235,11 @@ 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 = VaultKmsBackend::new(kms_config(config, Duration::from_secs(5), 3))
+            let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
                 .await
                 .expect("client construction performs no network calls");
-            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-definitely-missing"))
+            let error = client
+                .describe_key("fault-injection-definitely-missing", None)
                 .await
                 .expect_err("a missing key must resolve to key-not-found");
             assert!(matches!(error, KmsError::KeyNotFound { .. }), "got {error:?}");

crates/lifecycle/Cargo.toml

@@ -25,35 +25,7 @@ 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,21 +28,7 @@ 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,7 +540,6 @@ 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"));
@@ -627,7 +626,6 @@ impl DistributedLock {
     }
 
     /// Convenience: acquire exclusive lock as a guard
-    #[hotpath::measure]
     pub async fn lock_guard(
         &self,
         resource: ObjectKey,
@@ -642,7 +640,6 @@ impl DistributedLock {
     }
 
     /// Convenience: acquire exclusive lock with expected contention logs suppressed
-    #[hotpath::measure]
     pub async fn lock_guard_quiet(
         &self,
         resource: ObjectKey,
@@ -658,7 +655,6 @@ 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,14 +32,7 @@ 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,14 +28,7 @@ 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,40 +26,9 @@ 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,33 +34,7 @@ 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,7 +1035,6 @@ 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,
@@ -1143,7 +1142,6 @@ 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,7 +338,6 @@ 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,14 +27,7 @@ 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,50 +27,6 @@ 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.
@@ -102,7 +58,6 @@ 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,27 +30,9 @@ workspace = true
 
 [features]
 default = []
-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",
-]
+hotpath = ["hotpath/hotpath", "hotpath/reqwest-0-13"]
+hotpath-alloc = ["hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 
 [dependencies]
 rustfs-credentials = { workspace = true }

crates/protocols/Cargo.toml

@@ -30,55 +30,6 @@ 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",
@@ -111,7 +62,6 @@ 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,38 +32,7 @@ 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,14 +25,7 @@ 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,32 +28,7 @@ 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,32 +30,9 @@ workspace = true
 
 [features]
 default = []
-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",
-]
+hotpath = ["hotpath/hotpath", "hotpath/tokio", "hotpath/futures", "hotpath/reqwest-0-13"]
+hotpath-alloc = ["hotpath/hotpath-alloc"]
+hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 
 [dependencies]
 hotpath.workspace = true

crates/s3-ops/Cargo.toml

@@ -27,14 +27,7 @@ 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,14 +27,7 @@ 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,37 +28,7 @@ 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,20 +28,7 @@ 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,48 +29,7 @@ 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 }

crates/scanner/src/scanner.rs

@@ -2531,7 +2531,6 @@ where
 }
 
 #[instrument(skip_all)]
-#[hotpath::measure]
 async fn run_data_scanner_cycle(
     ctx: &CancellationToken,
     storeapi: &Arc<ECStore>,

crates/security-governance/Cargo.toml

@@ -30,12 +30,5 @@ 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,14 +25,7 @@ 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,14 +27,7 @@ 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/targets/Cargo.toml

@@ -11,41 +11,7 @@ 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,7 +522,6 @@ 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,
@@ -558,7 +557,6 @@ where
     Some((shared, cancel))
 }
 
-#[hotpath::measure]
 pub(crate) async fn prepare_target<E>(
     target: Box<dyn Target<E> + Send + Sync>,
     cancellation: Option<&CancellationToken>,
@@ -600,7 +598,6 @@ 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,
@@ -673,7 +670,6 @@ 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>,
@@ -809,7 +805,6 @@ 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,7 +85,6 @@ 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),
@@ -481,7 +480,6 @@ 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,32 +25,7 @@ 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,14 +27,7 @@ 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,20 +24,7 @@ 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,7 +227,6 @@ 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,7 +25,6 @@ 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 }
@@ -77,9 +76,6 @@ 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,14 +32,7 @@ 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/operations/hotpath-warp-abba-runbook.md

@@ -1,277 +0,0 @@
-# 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,137 +60,25 @@ 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,15 +2708,17 @@ 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);
 
-        self.router.insert(path, operation).map_err(std::io::Error::other)?;
-
         #[cfg(test)]
-        self.registered_routes.push(registered_path);
+        {
+            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)?;
 
         Ok(())
     }

scripts/run_hotpath_warp_abba.sh

@@ -1,428 +0,0 @@
-#!/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