mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-31 10:32:24 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 overtrue
|
5426243906 | ||
|
overtrue
|
c29946f82e | ||
|
overtrue
|
88155ff1a7 |
Generated
+32
-89
@@ -1528,7 +1528,7 @@ version = "0.10.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
|
||||
dependencies = [
|
||||
"generic-array 0.14.7",
|
||||
"generic-array 0.14.9",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1547,7 +1547,7 @@ version = "0.3.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93"
|
||||
dependencies = [
|
||||
"generic-array 0.14.7",
|
||||
"generic-array 0.14.9",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1598,7 +1598,7 @@ version = "3.9.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6dee98b0db6a962de883bf5d20362dee4d7ca0d12fe39a7c6c73c844e1cd7c1f"
|
||||
dependencies = [
|
||||
"darling 0.20.11",
|
||||
"darling 0.23.0",
|
||||
"ident_case",
|
||||
"prettyplease",
|
||||
"proc-macro2",
|
||||
@@ -1870,7 +1870,7 @@ version = "0.4.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
|
||||
dependencies = [
|
||||
"crypto-common 0.1.7",
|
||||
"crypto-common 0.1.6",
|
||||
"inout 0.1.4",
|
||||
]
|
||||
|
||||
@@ -2328,7 +2328,7 @@ version = "0.5.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76"
|
||||
dependencies = [
|
||||
"generic-array 0.14.7",
|
||||
"generic-array 0.14.9",
|
||||
"rand_core 0.6.4",
|
||||
"subtle",
|
||||
"zeroize",
|
||||
@@ -2353,11 +2353,11 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "crypto-common"
|
||||
version = "0.1.7"
|
||||
version = "0.1.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
|
||||
checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
|
||||
dependencies = [
|
||||
"generic-array 0.14.7",
|
||||
"generic-array 0.14.9",
|
||||
"typenum",
|
||||
]
|
||||
|
||||
@@ -3554,7 +3554,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
|
||||
dependencies = [
|
||||
"block-buffer 0.10.4",
|
||||
"const-oid 0.9.6",
|
||||
"crypto-common 0.1.7",
|
||||
"crypto-common 0.1.6",
|
||||
"subtle",
|
||||
]
|
||||
|
||||
@@ -3672,7 +3672,6 @@ dependencies = [
|
||||
"flate2",
|
||||
"futures",
|
||||
"hex",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
@@ -3814,7 +3813,7 @@ dependencies = [
|
||||
"crypto-bigint 0.5.5",
|
||||
"digest 0.10.7",
|
||||
"ff 0.13.1",
|
||||
"generic-array 0.14.7",
|
||||
"generic-array 0.14.9",
|
||||
"group 0.13.0",
|
||||
"hkdf 0.12.4",
|
||||
"pem-rfc7468 0.7.0",
|
||||
@@ -4259,9 +4258,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "generic-array"
|
||||
version = "0.14.7"
|
||||
version = "0.14.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
|
||||
checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2"
|
||||
dependencies = [
|
||||
"typenum",
|
||||
"version_check",
|
||||
@@ -4274,7 +4273,7 @@ version = "1.4.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ab4e5aa225bc56696909483320f0ff9b600f1a971b52e07a17d70f3d9b43254b"
|
||||
dependencies = [
|
||||
"generic-array 0.14.7",
|
||||
"generic-array 0.14.9",
|
||||
"rustversion",
|
||||
"typenum",
|
||||
]
|
||||
@@ -4373,9 +4372,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-auth"
|
||||
version = "1.15.0"
|
||||
version = "1.14.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f54aab44c16b8463ae11b165a87c3d484780231f157bb1ed65843d591beb5abd"
|
||||
checksum = "a3494870d06f3cbbb3561ada6f234982549e3a2fb31e719ef258e6eadb9ae09a"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"aws-lc-rs",
|
||||
@@ -4402,9 +4401,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-gax"
|
||||
version = "1.13.0"
|
||||
version = "1.12.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b9a46dd0fd026bbc4a5d84e6ab0c941cee6e3b057976a0bb107fdb5238ce598f"
|
||||
checksum = "3103a4a9013f1aed573ca56e19a9680b0211643a99ea85caf524b397d6be8be3"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures",
|
||||
@@ -4421,9 +4420,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-gax-internal"
|
||||
version = "0.7.16"
|
||||
version = "0.7.15"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fb04c54317ace06d489213f761797240b3046142a9b7ce6b9a82a9d134e193d1"
|
||||
checksum = "c0df265fba091ed7e00ecd0755009423310163f8b52820f007b6b4d97f4c6617"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures",
|
||||
@@ -4525,9 +4524,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-storage"
|
||||
version = "1.17.0"
|
||||
version = "1.16.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9227f65175fa91a6e41f246797917697efdadfe09dd8ea84ad8b737a71efbd28"
|
||||
checksum = "dc4b1d78c88db5c2530b12461e373a7d0d3a6caa3f6c1fc14e5d824cf2aeb307"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
@@ -4578,9 +4577,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "google-cloud-wkt"
|
||||
version = "1.7.0"
|
||||
version = "1.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7fccf98cfd5481a5f5a285181ab0c62123d7d47cd2bb7299448440649349e4e7"
|
||||
checksum = "46df1fcc3ab69164af3f4199ed21f45b5dbc56d9f03211eb4fa20116d442364b"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"bytes",
|
||||
@@ -4921,7 +4920,6 @@ dependencies = [
|
||||
"async-trait",
|
||||
"cfg-if",
|
||||
"crossbeam-channel",
|
||||
"flate2",
|
||||
"futures-channel",
|
||||
"futures-util",
|
||||
"hdrhistogram",
|
||||
@@ -4929,7 +4927,6 @@ dependencies = [
|
||||
"hotpath-meta",
|
||||
"http 1.5.0",
|
||||
"libc",
|
||||
"object 0.36.7",
|
||||
"parking_lot",
|
||||
"pin-project-lite",
|
||||
"prettytable-rs",
|
||||
@@ -4937,7 +4934,6 @@ dependencies = [
|
||||
"regex",
|
||||
"reqwest",
|
||||
"reqwest-middleware",
|
||||
"rustc-demangle",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tiny_http",
|
||||
@@ -5051,9 +5047,9 @@ checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15"
|
||||
|
||||
[[package]]
|
||||
name = "hybrid-array"
|
||||
version = "0.4.14"
|
||||
version = "0.4.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b"
|
||||
checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c"
|
||||
dependencies = [
|
||||
"ctutils",
|
||||
"subtle",
|
||||
@@ -5301,7 +5297,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01"
|
||||
dependencies = [
|
||||
"block-padding 0.3.3",
|
||||
"generic-array 0.14.7",
|
||||
"generic-array 0.14.9",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -6811,15 +6807,6 @@ dependencies = [
|
||||
"objc2-core-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "object"
|
||||
version = "0.36.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "object"
|
||||
version = "0.37.3"
|
||||
@@ -7857,7 +7844,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
|
||||
dependencies = [
|
||||
"heck",
|
||||
"itertools 0.13.0",
|
||||
"itertools 0.10.5",
|
||||
"log",
|
||||
"multimap",
|
||||
"once_cell",
|
||||
@@ -7877,7 +7864,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042"
|
||||
dependencies = [
|
||||
"heck",
|
||||
"itertools 0.13.0",
|
||||
"itertools 0.10.5",
|
||||
"log",
|
||||
"multimap",
|
||||
"petgraph 0.8.3",
|
||||
@@ -7898,7 +7885,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"itertools 0.13.0",
|
||||
"itertools 0.10.5",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
@@ -7911,7 +7898,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"itertools 0.13.0",
|
||||
"itertools 0.10.5",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
@@ -9040,7 +9027,6 @@ dependencies = [
|
||||
"const-str",
|
||||
"futures",
|
||||
"hashbrown 0.17.1",
|
||||
"hotpath",
|
||||
"metrics",
|
||||
"rustfs-config",
|
||||
"rustfs-s3-types",
|
||||
@@ -9061,7 +9047,6 @@ dependencies = [
|
||||
"base64-simd",
|
||||
"bytes",
|
||||
"crc-fast",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"md-5 0.11.0",
|
||||
"pretty_assertions",
|
||||
@@ -9075,7 +9060,6 @@ name = "rustfs-common"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"hotpath",
|
||||
"metrics",
|
||||
"rmp-serde",
|
||||
"s3s",
|
||||
@@ -9090,7 +9074,6 @@ dependencies = [
|
||||
name = "rustfs-concurrency"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"hotpath",
|
||||
"insta",
|
||||
"rustfs-io-core",
|
||||
"serde",
|
||||
@@ -9104,7 +9087,6 @@ name = "rustfs-config"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"const-str",
|
||||
"hotpath",
|
||||
"serde",
|
||||
"serde_json",
|
||||
]
|
||||
@@ -9115,7 +9097,6 @@ version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"base64-simd",
|
||||
"hmac 0.13.0",
|
||||
"hotpath",
|
||||
"rand 0.10.2",
|
||||
"serde",
|
||||
"serde_json",
|
||||
@@ -9131,7 +9112,6 @@ dependencies = [
|
||||
"argon2",
|
||||
"base64-simd",
|
||||
"chacha20poly1305",
|
||||
"hotpath",
|
||||
"jsonwebtoken 11.0.0",
|
||||
"pbkdf2 0.13.0",
|
||||
"rand 0.10.2",
|
||||
@@ -9149,7 +9129,6 @@ name = "rustfs-data-usage"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"hotpath",
|
||||
"rmp-serde",
|
||||
"rustfs-filemeta",
|
||||
"serde",
|
||||
@@ -9295,7 +9274,6 @@ dependencies = [
|
||||
name = "rustfs-extension-schema"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"hotpath",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"thiserror 2.0.19",
|
||||
@@ -9334,7 +9312,6 @@ dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.23.0",
|
||||
"futures",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"metrics",
|
||||
"rustfs-common",
|
||||
@@ -9366,7 +9343,6 @@ dependencies = [
|
||||
"async-trait",
|
||||
"base64-simd",
|
||||
"futures",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"jsonwebtoken 11.0.0",
|
||||
"moka",
|
||||
@@ -9401,7 +9377,6 @@ name = "rustfs-io-core"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"hotpath",
|
||||
"memmap2",
|
||||
"rustfs-io-metrics",
|
||||
"thiserror 2.0.19",
|
||||
@@ -9414,7 +9389,6 @@ name = "rustfs-io-metrics"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"criterion",
|
||||
"hotpath",
|
||||
"metrics",
|
||||
"metrics-util",
|
||||
"num_cpus",
|
||||
@@ -9481,7 +9455,6 @@ version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"http-body 1.1.0",
|
||||
"http-body-util",
|
||||
@@ -9514,12 +9487,9 @@ dependencies = [
|
||||
"base64 0.23.0",
|
||||
"chacha20poly1305",
|
||||
"hex",
|
||||
"hotpath",
|
||||
"insta",
|
||||
"jiff",
|
||||
"md-5 0.11.0",
|
||||
"metrics",
|
||||
"metrics-util",
|
||||
"moka",
|
||||
"rand 0.10.2",
|
||||
"reqwest",
|
||||
@@ -9547,7 +9517,6 @@ name = "rustfs-lifecycle"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"hotpath",
|
||||
"metrics",
|
||||
"metrics-util",
|
||||
"proptest",
|
||||
@@ -9572,7 +9541,6 @@ dependencies = [
|
||||
"async-trait",
|
||||
"crossbeam-queue",
|
||||
"futures",
|
||||
"hotpath",
|
||||
"parking_lot",
|
||||
"rand 0.10.2",
|
||||
"rustfs-io-metrics",
|
||||
@@ -9594,7 +9562,6 @@ version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"flate2",
|
||||
"hotpath",
|
||||
"regex",
|
||||
"serde",
|
||||
"serde_json",
|
||||
@@ -9612,7 +9579,6 @@ name = "rustfs-madmin"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"hotpath",
|
||||
"humantime",
|
||||
"hyper",
|
||||
"rmp-serde",
|
||||
@@ -9633,7 +9599,6 @@ dependencies = [
|
||||
"criterion",
|
||||
"form_urlencoded",
|
||||
"hashbrown 0.17.1",
|
||||
"hotpath",
|
||||
"metrics",
|
||||
"percent-encoding",
|
||||
"quick-xml",
|
||||
@@ -9663,7 +9628,6 @@ version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"criterion",
|
||||
"futures",
|
||||
"hotpath",
|
||||
"rustfs-config",
|
||||
"rustfs-io-metrics",
|
||||
"rustfs-utils",
|
||||
@@ -9682,7 +9646,6 @@ version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"criterion",
|
||||
"hotpath",
|
||||
"metrics",
|
||||
"metrics-util",
|
||||
"moka",
|
||||
@@ -9705,7 +9668,6 @@ dependencies = [
|
||||
"flate2",
|
||||
"futures-util",
|
||||
"glob",
|
||||
"hotpath",
|
||||
"jiff",
|
||||
"libc",
|
||||
"metrics",
|
||||
@@ -9754,7 +9716,6 @@ dependencies = [
|
||||
"base64-simd",
|
||||
"chrono",
|
||||
"futures",
|
||||
"hotpath",
|
||||
"ipnetwork",
|
||||
"jsonwebtoken 11.0.0",
|
||||
"moka",
|
||||
@@ -9791,7 +9752,6 @@ dependencies = [
|
||||
"futures-util",
|
||||
"hex",
|
||||
"hmac 0.13.0",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
@@ -9844,7 +9804,6 @@ name = "rustfs-protos"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"flatbuffers",
|
||||
"hotpath",
|
||||
"prost 0.14.4",
|
||||
"rmp-serde",
|
||||
"rustfs-common",
|
||||
@@ -9869,7 +9828,6 @@ version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"byteorder",
|
||||
"bytes",
|
||||
"hotpath",
|
||||
"regex",
|
||||
"rmp",
|
||||
"rmp-serde",
|
||||
@@ -9928,7 +9886,6 @@ dependencies = [
|
||||
"chacha20poly1305",
|
||||
"hex",
|
||||
"hmac 0.13.0",
|
||||
"hotpath",
|
||||
"minlz",
|
||||
"pin-project-lite",
|
||||
"rand 0.10.2",
|
||||
@@ -9946,7 +9903,6 @@ dependencies = [
|
||||
name = "rustfs-s3-ops"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"hotpath",
|
||||
"rustfs-s3-types",
|
||||
]
|
||||
|
||||
@@ -9954,7 +9910,6 @@ dependencies = [
|
||||
name = "rustfs-s3-types"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"hotpath",
|
||||
"serde",
|
||||
"serde_json",
|
||||
]
|
||||
@@ -9969,7 +9924,6 @@ dependencies = [
|
||||
"datafusion",
|
||||
"futures",
|
||||
"futures-core",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"metrics",
|
||||
"parking_lot",
|
||||
@@ -9998,7 +9952,6 @@ dependencies = [
|
||||
"datafusion",
|
||||
"derive_builder",
|
||||
"futures",
|
||||
"hotpath",
|
||||
"parking_lot",
|
||||
"rustfs-s3select-api",
|
||||
"s3s",
|
||||
@@ -10016,7 +9969,6 @@ dependencies = [
|
||||
"futures",
|
||||
"hex-simd",
|
||||
"hmac 0.13.0",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"metrics",
|
||||
"rand 0.10.2",
|
||||
@@ -10049,7 +10001,6 @@ dependencies = [
|
||||
name = "rustfs-security-governance"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"hotpath",
|
||||
"thiserror 2.0.19",
|
||||
]
|
||||
|
||||
@@ -10059,7 +10010,6 @@ version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"base64-simd",
|
||||
"bytes",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"hyper",
|
||||
"rustfs-utils",
|
||||
@@ -10076,7 +10026,6 @@ name = "rustfs-storage-api"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"hotpath",
|
||||
"insta",
|
||||
"rustfs-filemeta",
|
||||
"serde",
|
||||
@@ -10098,7 +10047,6 @@ dependencies = [
|
||||
"deadpool-postgres",
|
||||
"futures-util",
|
||||
"hashbrown 0.17.1",
|
||||
"hotpath",
|
||||
"hyper",
|
||||
"hyper-rustls",
|
||||
"lapin",
|
||||
@@ -10144,7 +10092,6 @@ dependencies = [
|
||||
name = "rustfs-test-utils"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"hotpath",
|
||||
"rustfs-data-usage",
|
||||
"rustfs-ecstore",
|
||||
"rustfs-storage-api",
|
||||
@@ -10161,7 +10108,6 @@ name = "rustfs-tls-runtime"
|
||||
version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"hotpath",
|
||||
"metrics",
|
||||
"rcgen",
|
||||
"rustfs-common",
|
||||
@@ -10183,7 +10129,6 @@ version = "1.0.0-beta.12"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"axum",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"ipnetwork",
|
||||
"metrics",
|
||||
@@ -10230,7 +10175,6 @@ dependencies = [
|
||||
"hex-simd",
|
||||
"highway",
|
||||
"hmac 0.13.0",
|
||||
"hotpath",
|
||||
"http 1.5.0",
|
||||
"hyper",
|
||||
"local-ip-address",
|
||||
@@ -10263,7 +10207,6 @@ dependencies = [
|
||||
"astral-tokio-tar",
|
||||
"async-compression",
|
||||
"criterion",
|
||||
"hotpath",
|
||||
"tempfile",
|
||||
"thiserror 2.0.19",
|
||||
"tokio",
|
||||
@@ -10607,7 +10550,7 @@ checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc"
|
||||
dependencies = [
|
||||
"base16ct 0.2.0",
|
||||
"der 0.7.10",
|
||||
"generic-array 0.14.7",
|
||||
"generic-array 0.14.9",
|
||||
"pkcs8 0.10.2",
|
||||
"subtle",
|
||||
"zeroize",
|
||||
@@ -11597,7 +11540,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
|
||||
dependencies = [
|
||||
"fastrand",
|
||||
"getrandom 0.4.3",
|
||||
"getrandom 0.3.4",
|
||||
"once_cell",
|
||||
"rustix",
|
||||
"windows-sys 0.61.2",
|
||||
|
||||
+2
-2
@@ -250,8 +250,8 @@ enumset = "1.1.14"
|
||||
faster-hex = "0.10.0"
|
||||
flate2 = "1.1.9"
|
||||
glob = "0.3.4"
|
||||
google-cloud-storage = "1.17.0"
|
||||
google-cloud-auth = "1.15.0"
|
||||
google-cloud-storage = "1.16.0"
|
||||
google-cloud-auth = "1.14.0"
|
||||
hashbrown = { version = "0.17.1" }
|
||||
hex = "0.4.3"
|
||||
hex-simd = "0.8.0"
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -40,83 +40,13 @@ 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`.
|
||||
|
||||
@@ -579,26 +579,8 @@ pub struct BucketMetadataSys {
|
||||
/// Serializes metadata-map commits and their derived cache updates for one
|
||||
/// bucket. Namespace locks, when present, are acquired before this lock.
|
||||
metadata_publish_locks: Arc<MetadataPublishLockRegistry>,
|
||||
/// Deduplicates concurrent lazy loads of one bucket's metadata, so N
|
||||
/// simultaneous cache misses issue a single disk read instead of N.
|
||||
///
|
||||
/// This is the `singleflight` that upstream applies to its own lazy
|
||||
/// `GetConfig`. Without it the namespace *read* lock the load holds is no
|
||||
/// help: read locks are shared, so it excludes concurrent config writers
|
||||
/// but not concurrent readers, and every caller still pays a full
|
||||
/// erasure-set metadata fanout. A separate registry from
|
||||
/// `metadata_publish_locks`, reusing the same per-bucket lock machinery.
|
||||
///
|
||||
/// Lock order: this lock, then the namespace lock, then the publish lock,
|
||||
/// then the metadata map. It is only ever taken as the first of those, so
|
||||
/// it cannot invert against a path that already holds one of the others.
|
||||
lazy_load_locks: Arc<MetadataPublishLockRegistry>,
|
||||
#[cfg(test)]
|
||||
lazy_load_lock_probe: std::sync::atomic::AtomicBool,
|
||||
/// Counts disk loads taken by the lazy `get_config` path, so a test can
|
||||
/// prove concurrent misses collapse into one.
|
||||
#[cfg(test)]
|
||||
lazy_disk_loads: std::sync::atomic::AtomicUsize,
|
||||
/// Buckets recently observed to have no persisted metadata. Serving the
|
||||
/// fabricated default from here (instead of re-reading disk) keeps the
|
||||
/// per-request cost of repeated lookups for such names bounded — without
|
||||
@@ -617,13 +599,8 @@ impl BucketMetadataSys {
|
||||
metadata_publish_locks: Arc::new(MetadataPublishLockRegistry {
|
||||
locks: StdMutex::new(HashMap::new()),
|
||||
}),
|
||||
lazy_load_locks: Arc::new(MetadataPublishLockRegistry {
|
||||
locks: StdMutex::new(HashMap::new()),
|
||||
}),
|
||||
#[cfg(test)]
|
||||
lazy_load_lock_probe: std::sync::atomic::AtomicBool::new(false),
|
||||
#[cfg(test)]
|
||||
lazy_disk_loads: std::sync::atomic::AtomicUsize::new(0),
|
||||
absent_metadata: moka::future::Cache::builder()
|
||||
.max_capacity(ABSENT_BUCKET_METADATA_MAX_ENTRIES)
|
||||
.time_to_live(ABSENT_BUCKET_METADATA_TTL)
|
||||
@@ -638,22 +615,16 @@ impl BucketMetadataSys {
|
||||
}
|
||||
|
||||
fn metadata_publish_lock(&self, bucket: &str) -> Arc<Mutex<MetadataPublishLockState>> {
|
||||
Self::bucket_lock_in(&self.metadata_publish_locks, bucket)
|
||||
}
|
||||
|
||||
/// Per-bucket gate for the lazy `get_config` disk load. See
|
||||
/// [`Self::lazy_load_locks`].
|
||||
fn lazy_load_lock(&self, bucket: &str) -> Arc<Mutex<MetadataPublishLockState>> {
|
||||
Self::bucket_lock_in(&self.lazy_load_locks, bucket)
|
||||
}
|
||||
|
||||
fn bucket_lock_in(registry: &Arc<MetadataPublishLockRegistry>, bucket: &str) -> Arc<Mutex<MetadataPublishLockState>> {
|
||||
let mut locks = registry.locks.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
let mut locks = self
|
||||
.metadata_publish_locks
|
||||
.locks
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
locks.get(bucket).and_then(Weak::upgrade).unwrap_or_else(|| {
|
||||
let lock = Arc::new_cyclic(|lock| {
|
||||
Mutex::new(MetadataPublishLockState {
|
||||
bucket: bucket.to_string(),
|
||||
registry: Arc::downgrade(registry),
|
||||
registry: Arc::downgrade(&self.metadata_publish_locks),
|
||||
lock: lock.clone(),
|
||||
})
|
||||
});
|
||||
@@ -1108,27 +1079,6 @@ impl BucketMetadataSys {
|
||||
return Ok((Arc::new(bm), true));
|
||||
}
|
||||
|
||||
// Collapse concurrent misses for this bucket into one disk load.
|
||||
// Taken before the namespace lock — see `lazy_load_locks` for the
|
||||
// ordering rule.
|
||||
let load_lock = self.lazy_load_lock(bucket);
|
||||
let _load_guard = load_lock.lock_owned().await;
|
||||
|
||||
// Re-check both caches: whoever held the gate before us may have
|
||||
// already answered this exact question, and repeating the fanout
|
||||
// is the whole cost this gate exists to avoid.
|
||||
if let Some(bm) = self.metadata_map.read().await.get(bucket).cloned() {
|
||||
return Ok((bm, true));
|
||||
}
|
||||
if self.absent_metadata.get(bucket).await.is_some() {
|
||||
let mut bm = BucketMetadata::new(bucket);
|
||||
bm.default_timestamps();
|
||||
return Ok((Arc::new(bm), true));
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
self.lazy_disk_loads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
let lock = self.api.new_ns_lock(bucket, bucket).await?;
|
||||
let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
|
||||
#[cfg(test)]
|
||||
@@ -1481,43 +1431,6 @@ mod tests {
|
||||
use serial_test::serial;
|
||||
use tokio::time::timeout;
|
||||
|
||||
/// Concurrent cache misses for one bucket must collapse into a single disk
|
||||
/// load.
|
||||
///
|
||||
/// The namespace read lock the lazy path already holds does not provide
|
||||
/// this: read locks are shared, so it excludes concurrent config writers
|
||||
/// but not concurrent readers. Without the dedup gate every caller pays its
|
||||
/// own namespace-lock acquisition plus a full erasure-set metadata fanout —
|
||||
/// and the paths that reach `get_config` are per-request, so the multiplier
|
||||
/// is request concurrency.
|
||||
#[tokio::test]
|
||||
async fn concurrent_lazy_loads_of_one_bucket_issue_a_single_disk_read() {
|
||||
use std::sync::atomic::Ordering;
|
||||
|
||||
let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
|
||||
let sys = Arc::new(BucketMetadataSys::new(ecstore));
|
||||
|
||||
// A name with no persisted metadata: every caller misses the map, and
|
||||
// the absent-cache entry does not exist until the first load records it.
|
||||
let bucket = "singleflight-bucket";
|
||||
let waiters = 8;
|
||||
|
||||
let results = futures::future::join_all((0..waiters).map(|_| {
|
||||
let sys = Arc::clone(&sys);
|
||||
async move { sys.get_config(bucket).await.map(|(bm, _)| bm.name.clone()) }
|
||||
}))
|
||||
.await;
|
||||
|
||||
for result in results {
|
||||
assert_eq!(result.expect("every caller must get an answer"), bucket);
|
||||
}
|
||||
assert_eq!(
|
||||
sys.lazy_disk_loads.load(Ordering::Relaxed),
|
||||
1,
|
||||
"concurrent misses for one bucket must share a single disk load"
|
||||
);
|
||||
}
|
||||
|
||||
/// Pins the fail-closed caching contract of the lazy `get_config` path
|
||||
/// and the refresh no-replace rule: fabricated defaults are returned but
|
||||
/// never served by the map-only `get()`, persisted metadata is cached on
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -27,9 +27,8 @@ 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"]
|
||||
|
||||
[dependencies]
|
||||
hotpath.workspace = true
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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());
|
||||
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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?;
|
||||
|
||||
|
||||
@@ -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"] }
|
||||
@@ -38,8 +37,6 @@ serde = { workspace = true, features = ["derive"] }
|
||||
serde_json = { workspace = true, features = ["raw_value"] }
|
||||
tracing = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
# Operation metrics emitted by the retry policy engine (crate::policy).
|
||||
metrics = { workspace = true }
|
||||
|
||||
# Cryptography
|
||||
aes-gcm = { workspace = true, features = ["rand_core"] }
|
||||
@@ -75,8 +72,6 @@ tokio-util = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
anyhow = { workspace = true }
|
||||
# Debugging recorder for asserting emitted metrics in tests.
|
||||
metrics-util = { version = "0.20", features = ["debugging"] }
|
||||
insta = { workspace = true, features = ["yaml", "json"] }
|
||||
tempfile = { workspace = true }
|
||||
temp-env = { workspace = true }
|
||||
@@ -85,17 +80,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"]
|
||||
|
||||
@@ -399,6 +399,20 @@ pub struct LocalKmsClient {
|
||||
/// Per-key write locks serializing read-modify-write updates within this
|
||||
/// process (see [`Self::lock_key_for_write`]).
|
||||
key_write_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
|
||||
/// Directory-wide writer fence for backup export (see
|
||||
/// [`Self::acquire_export_fence`]). Writers hold the read side; an export
|
||||
/// snapshot holds the write side so it observes a single-generation view.
|
||||
export_fence: Arc<tokio::sync::RwLock<()>>,
|
||||
}
|
||||
|
||||
/// Guard pairing the export-fence read lock with a per-key write mutex.
|
||||
///
|
||||
/// Dropping it releases both, so every existing `lock_key_for_write` call
|
||||
/// site participates in the export fence without changes.
|
||||
#[must_use]
|
||||
struct KeyWriteGuard {
|
||||
_fence: tokio::sync::OwnedRwLockReadGuard<()>,
|
||||
_key: tokio::sync::OwnedMutexGuard<()>,
|
||||
}
|
||||
|
||||
// pub(crate) so the backup contract tests can anchor the manifest's
|
||||
@@ -465,6 +479,7 @@ impl LocalKmsClient {
|
||||
legacy_master_cipher,
|
||||
dek_crypto: AesDekCrypto::new(),
|
||||
key_write_locks: Mutex::new(HashMap::new()),
|
||||
export_fence: Arc::new(tokio::sync::RwLock::new(())),
|
||||
};
|
||||
client.validate_existing_keys().await?;
|
||||
Ok(client)
|
||||
@@ -507,6 +522,7 @@ impl LocalKmsClient {
|
||||
legacy_master_cipher,
|
||||
dek_crypto: AesDekCrypto::new(),
|
||||
key_write_locks: Mutex::new(HashMap::new()),
|
||||
export_fence: Arc::new(tokio::sync::RwLock::new(())),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -517,12 +533,40 @@ impl LocalKmsClient {
|
||||
/// delete with a rewrite. Cross-process writers sharing a key directory
|
||||
/// remain unsupported. Entries live for the client's lifetime; the table
|
||||
/// is bounded by the number of distinct key ids this process touches.
|
||||
async fn lock_key_for_write(&self, key_id: &str) -> tokio::sync::OwnedMutexGuard<()> {
|
||||
async fn lock_key_for_write(&self, key_id: &str) -> KeyWriteGuard {
|
||||
// Fence first, per-key mutex second: the ordering is uniform across
|
||||
// all writers, so an export waiting on the write side can never
|
||||
// deadlock with a writer holding a key mutex.
|
||||
let fence = Arc::clone(&self.export_fence).read_owned().await;
|
||||
let lock = {
|
||||
let mut locks = self.key_write_locks.lock().expect("Local KMS key write lock table poisoned");
|
||||
Arc::clone(locks.entry(key_id.to_string()).or_default())
|
||||
};
|
||||
lock.lock_owned().await
|
||||
KeyWriteGuard {
|
||||
_fence: fence,
|
||||
_key: lock.lock_owned().await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Block every key-directory writer while a backup export collects its
|
||||
/// snapshot, so all records belong to one generation.
|
||||
///
|
||||
/// Mutating operations hold the read side (via [`Self::lock_key_for_write`]
|
||||
/// or [`Self::save_new_master_key`]); the export holds the write side only
|
||||
/// for the collection phase, never while encrypting or writing the bundle.
|
||||
pub(crate) async fn acquire_export_fence(&self) -> tokio::sync::OwnedRwLockWriteGuard<()> {
|
||||
Arc::clone(&self.export_fence).write_owned().await
|
||||
}
|
||||
|
||||
/// Key directory root, exposed for the backup export module.
|
||||
pub(crate) fn key_directory(&self) -> &Path {
|
||||
&self.config.key_dir
|
||||
}
|
||||
|
||||
/// Absolute path of the master-key KDF salt file, exposed for the backup
|
||||
/// export module.
|
||||
pub(crate) fn master_key_salt_file(&self) -> PathBuf {
|
||||
Self::master_key_salt_path(&self.config)
|
||||
}
|
||||
|
||||
/// Derive a 256-bit key from the master key string using a persistent Argon2id salt.
|
||||
@@ -799,6 +843,11 @@ impl LocalKmsClient {
|
||||
}
|
||||
|
||||
async fn save_new_master_key(&self, master_key: &MasterKeyInfo, key_material: &[u8]) -> Result<()> {
|
||||
// Creates never take the per-key write lock (`NoClobber` publishing
|
||||
// already linearizes them), so they join the export fence here. This
|
||||
// must stay the only fence acquisition on the create path: the fence
|
||||
// read lock is not reentrant while an export waits for the write side.
|
||||
let _fence = Arc::clone(&self.export_fence).read_owned().await;
|
||||
let key_path = self.master_key_path(&master_key.key_id)?;
|
||||
let content = self.encode_master_key(master_key, key_material)?;
|
||||
let temp_path = key_path.with_extension(format!("tmp-{}", uuid::Uuid::new_v4()));
|
||||
|
||||
@@ -0,0 +1,999 @@
|
||||
// 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::backends::KmsClient;
|
||||
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:?}");
|
||||
}
|
||||
}
|
||||
@@ -355,8 +355,10 @@ struct ManifestProbe {
|
||||
///
|
||||
/// The manifest is the authoritative description of one backup bundle: what
|
||||
/// was captured, under which snapshot generation, protected by which backup
|
||||
/// KEK, and which restore responsibility applies. Field order is part of the
|
||||
/// canonical digest form and is frozen for this format version.
|
||||
/// KEK, and which restore responsibility applies. The digest's canonical
|
||||
/// form is the manifest's JSON value with the digest hex emptied (see
|
||||
/// [`Self::compute_digest`]); decoders verify it against the raw stored
|
||||
/// bytes and never re-serialize parsed fields.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct BackupManifest {
|
||||
@@ -415,8 +417,14 @@ impl BackupManifest {
|
||||
///
|
||||
/// Fail-closed order: truncated or malformed input, then unknown format
|
||||
/// version, then a missing completeness marker, then schema decoding
|
||||
/// (unknown fields, duplicate fields, missing fields), then semantic
|
||||
/// validation including digest verification.
|
||||
/// (unknown fields, duplicate fields, missing fields), then digest
|
||||
/// verification against the raw input bytes, then semantic validation.
|
||||
///
|
||||
/// The digest is verified against the bytes as stored — parsed typed
|
||||
/// fields are never re-serialized for verification, so a field whose
|
||||
/// string form does not round-trip byte-identically through its parsed
|
||||
/// representation (timestamps in environment-dependent time zone
|
||||
/// spellings, for example) cannot produce a spurious mismatch.
|
||||
pub fn decode(bytes: &[u8]) -> Result<Self, BackupError> {
|
||||
let probe: ManifestProbe = serde_json::from_slice(bytes).map_err(map_serde_error)?;
|
||||
if probe.format_version != Self::FORMAT_VERSION {
|
||||
@@ -429,7 +437,9 @@ impl BackupManifest {
|
||||
return Err(BackupError::incomplete_bundle("manifest has no completeness marker"));
|
||||
}
|
||||
let manifest: Self = serde_json::from_slice(bytes).map_err(map_serde_error)?;
|
||||
manifest.validate()?;
|
||||
manifest.validate_pre_digest()?;
|
||||
Self::verify_digest_in_bytes(bytes, &manifest.manifest_digest)?;
|
||||
manifest.validate_content()?;
|
||||
Ok(manifest)
|
||||
}
|
||||
|
||||
@@ -449,23 +459,30 @@ impl BackupManifest {
|
||||
Ok(self)
|
||||
}
|
||||
|
||||
/// Compute the digest over the canonical manifest bytes.
|
||||
/// Compute the digest over the canonical manifest form.
|
||||
///
|
||||
/// Canonical form: compact JSON serialization of this manifest with the
|
||||
/// digest hex emptied. Field order is struct declaration order and is
|
||||
/// frozen for format version 1, so the same manifest content always
|
||||
/// hashes to the same value.
|
||||
/// Canonical form: the JSON *value* of the manifest with the digest hex
|
||||
/// emptied, object keys rebuilt in bytewise-sorted order at every level
|
||||
/// (see [`canonicalize_value`]), then serialized compactly. The value
|
||||
/// layer is what makes sealing and decoding agree byte-for-byte — a
|
||||
/// decoder recovers the identical value from the raw stored bytes
|
||||
/// without round-tripping any typed field through parse-and-reprint —
|
||||
/// and the explicit key sort makes the bytes independent of
|
||||
/// `serde_json`'s map implementation (`preserve_order` on or off).
|
||||
pub fn compute_digest(&self) -> Result<ContentDigest, BackupError> {
|
||||
let mut unsealed = self.clone();
|
||||
unsealed.manifest_digest = ContentDigest::placeholder(self.manifest_digest.algorithm);
|
||||
let canonical = serde_json::to_vec(&unsealed)
|
||||
let value = serde_json::to_value(&unsealed)
|
||||
.map_err(|error| BackupError::corrupted(format!("manifest canonicalization failed: {error}")))?;
|
||||
match self.manifest_digest.algorithm {
|
||||
DigestAlgorithm::Sha256 => Ok(ContentDigest::sha256_of(&canonical)),
|
||||
}
|
||||
Self::digest_of_canonical_value(value, self.manifest_digest.algorithm)
|
||||
}
|
||||
|
||||
/// Verify the sealed digest against the current manifest content.
|
||||
/// Verify the sealed digest against the current in-memory content.
|
||||
///
|
||||
/// This is the producer-side check (sealing and [`Self::encode`]).
|
||||
/// Decoders must use the raw stored bytes instead (see [`Self::decode`]):
|
||||
/// re-serializing parsed fields is not guaranteed to reproduce the
|
||||
/// stored spelling byte-for-byte.
|
||||
pub fn verify_digest(&self) -> Result<(), BackupError> {
|
||||
if !self.manifest_digest.is_well_formed() {
|
||||
return Err(BackupError::corrupted("manifest digest is not a well-formed digest value"));
|
||||
@@ -478,6 +495,33 @@ impl BackupManifest {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Verify a declared digest against raw manifest bytes, normalizing only
|
||||
/// through the JSON value layer and emptying the digest slot in place.
|
||||
fn verify_digest_in_bytes(bytes: &[u8], declared: &ContentDigest) -> Result<(), BackupError> {
|
||||
if !declared.is_well_formed() {
|
||||
return Err(BackupError::corrupted("manifest digest is not a well-formed digest value"));
|
||||
}
|
||||
let mut value: serde_json::Value = serde_json::from_slice(bytes).map_err(map_serde_error)?;
|
||||
let Some(slot) = value.get_mut("manifest_digest").and_then(|digest| digest.get_mut("hex")) else {
|
||||
return Err(BackupError::corrupted("manifest has no digest slot"));
|
||||
};
|
||||
*slot = serde_json::Value::String(String::new());
|
||||
if Self::digest_of_canonical_value(value, declared.algorithm)? != *declared {
|
||||
return Err(BackupError::corrupted(
|
||||
"manifest digest mismatch: content does not match the sealed digest",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn digest_of_canonical_value(value: serde_json::Value, algorithm: DigestAlgorithm) -> Result<ContentDigest, BackupError> {
|
||||
let canonical = serde_json::to_vec(&canonicalize_value(value))
|
||||
.map_err(|error| BackupError::corrupted(format!("manifest canonicalization failed: {error}")))?;
|
||||
match algorithm {
|
||||
DigestAlgorithm::Sha256 => Ok(ContentDigest::sha256_of(&canonical)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Look up a required artifact by kind, failing closed when absent.
|
||||
pub fn require_artifact(&self, kind: ArtifactKind) -> Result<&ArtifactDescriptor, BackupError> {
|
||||
self.artifacts
|
||||
@@ -486,11 +530,21 @@ impl BackupManifest {
|
||||
.ok_or_else(|| BackupError::missing_artifact(artifact_kind_name(kind)))
|
||||
}
|
||||
|
||||
/// Validate the full manifest contract.
|
||||
/// Validate the full manifest contract against the in-memory content.
|
||||
///
|
||||
/// This is decode-side validation and also guards [`Self::encode`], so a
|
||||
/// producer cannot publish a manifest a decoder would reject.
|
||||
/// This guards [`Self::encode`], so a producer cannot publish a manifest
|
||||
/// a decoder would reject. [`Self::decode`] runs the same checks but
|
||||
/// verifies the digest against the raw input bytes instead.
|
||||
pub fn validate(&self) -> Result<(), BackupError> {
|
||||
self.validate_pre_digest()?;
|
||||
self.verify_digest()?;
|
||||
self.validate_content()
|
||||
}
|
||||
|
||||
/// Checks that must run before any digest verification: an unknown
|
||||
/// version or an unsealed bundle is reported as its own typed error, not
|
||||
/// as a digest mismatch.
|
||||
fn validate_pre_digest(&self) -> Result<(), BackupError> {
|
||||
if self.format_version != Self::FORMAT_VERSION {
|
||||
return Err(BackupError::UnknownVersion {
|
||||
found: self.format_version,
|
||||
@@ -500,7 +554,12 @@ impl BackupManifest {
|
||||
if self.completeness != CompletenessState::Complete {
|
||||
return Err(BackupError::incomplete_bundle("completeness marker records an in-progress bundle"));
|
||||
}
|
||||
self.verify_digest()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Semantic validation of everything except version, completeness, and
|
||||
/// digest integrity.
|
||||
fn validate_content(&self) -> Result<(), BackupError> {
|
||||
require_non_empty("backup_id", &self.backup_id)?;
|
||||
require_non_empty("rustfs_version", &self.rustfs_version)?;
|
||||
require_non_empty("deployment_identity", &self.deployment_identity)?;
|
||||
@@ -645,6 +704,29 @@ fn map_serde_error(error: serde_json::Error) -> BackupError {
|
||||
}
|
||||
}
|
||||
|
||||
/// Rebuild a JSON value with object keys in bytewise-sorted order at every
|
||||
/// nesting level (array element order is preserved).
|
||||
///
|
||||
/// `serde_json`'s map keeps keys sorted by default but preserves insertion
|
||||
/// order when the `preserve_order` feature is unified into the build by any
|
||||
/// other crate. Digest bytes must not depend on that, so the ordering is
|
||||
/// imposed explicitly here instead of being inherited from the map type.
|
||||
fn canonicalize_value(value: serde_json::Value) -> serde_json::Value {
|
||||
match value {
|
||||
serde_json::Value::Object(map) => {
|
||||
let mut entries: Vec<(String, serde_json::Value)> = map.into_iter().collect();
|
||||
entries.sort_by(|a, b| a.0.cmp(&b.0));
|
||||
let mut sorted = serde_json::Map::with_capacity(entries.len());
|
||||
for (key, entry) in entries {
|
||||
sorted.insert(key, canonicalize_value(entry));
|
||||
}
|
||||
serde_json::Value::Object(sorted)
|
||||
}
|
||||
serde_json::Value::Array(items) => serde_json::Value::Array(items.into_iter().map(canonicalize_value).collect()),
|
||||
other => other,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -736,7 +818,7 @@ mod tests {
|
||||
/// canonical form and frozen. If serialization layout or field order
|
||||
/// changes, this value changes and the fixture test fails — which is the
|
||||
/// point: that is a format-version bump, not a patch.
|
||||
const FIXTURE_DIGEST_HEX: &str = "01accb3e2bc51e12d17d1efc52ad6ab9441c50bec52c3fb29e0a9f92b725cdaa";
|
||||
const FIXTURE_DIGEST_HEX: &str = "a8b104d61c358cd75cc9a7691d2f7d2d96f73c53653bef8940a49e96dbdf7775";
|
||||
|
||||
fn fixture() -> String {
|
||||
FIXTURE.replace("SEALED_DIGEST_HEX", FIXTURE_DIGEST_HEX)
|
||||
@@ -1010,12 +1092,39 @@ mod tests {
|
||||
let with_discovery = fixture().replace("\"completeness\"", "\"capability_discovery\": [1], \"completeness\"");
|
||||
expect_corrupted(BackupManifest::decode(with_discovery.as_bytes()), "reserved");
|
||||
|
||||
// Explicit null carries no data and is tolerated as absence; digest
|
||||
// verification still passes because null slots are skipped on
|
||||
// serialization.
|
||||
// An explicit null slot decodes as absence at the schema layer, but
|
||||
// sealed bundles never contain the key (`skip_serializing_if`), so
|
||||
// inserting one after sealing is a byte-level modification and the
|
||||
// raw-bytes digest check rejects it.
|
||||
let with_null = fixture().replace("\"completeness\"", "\"key_versions\": null, \"completeness\"");
|
||||
let decoded = BackupManifest::decode(with_null.as_bytes()).expect("null reserved slot should decode");
|
||||
assert_eq!(decoded.key_versions, None);
|
||||
expect_corrupted(BackupManifest::decode(with_null.as_bytes()), "digest mismatch");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn digest_verification_survives_non_round_tripping_timestamp_spellings() {
|
||||
// A legacy `created_at` spelling (no time zone annotation) parses via
|
||||
// the compat fallback and re-serializes differently ("+00:00[UTC]"),
|
||||
// and host-dependent zone spellings can do the same. Digest
|
||||
// verification therefore operates on the raw stored bytes and must
|
||||
// never re-serialize parsed fields.
|
||||
let sealed = seal(local_manifest_unsealed());
|
||||
let mut value = serde_json::to_value(&sealed).expect("manifest should convert to a value");
|
||||
value["created_at"] = serde_json::Value::String("2026-07-30T00:00:00+00:00".to_string());
|
||||
value["manifest_digest"]["hex"] = serde_json::Value::String(String::new());
|
||||
let digest = BackupManifest::digest_of_canonical_value(value.clone(), DigestAlgorithm::Sha256)
|
||||
.expect("canonical digest should compute");
|
||||
value["manifest_digest"]["hex"] = serde_json::Value::String(digest.hex);
|
||||
let bytes = serde_json::to_vec(&value).expect("manifest bytes");
|
||||
|
||||
let decoded = BackupManifest::decode(&bytes).expect("a non-round-tripping timestamp spelling must not break decoding");
|
||||
|
||||
// Precondition: the spelling really does not survive a typed
|
||||
// round-trip — otherwise this test is vacuous.
|
||||
let reserialized = serde_json::to_value(&decoded).expect("decoded manifest should convert to a value");
|
||||
assert_ne!(reserialized["created_at"], value["created_at"]);
|
||||
// Which is exactly why the producer-side (in-memory) digest check
|
||||
// cannot be used on decoded manifests.
|
||||
assert!(decoded.verify_digest().is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -12,13 +12,13 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Backup/restore contract types for KMS state.
|
||||
//! Backup/restore contracts and backup production for KMS state.
|
||||
//!
|
||||
//! This module is contract-only: it defines the versioned backup manifest,
|
||||
//! the per-backend responsibility matrix, typed failure modes, and the
|
||||
//! restore dry-run report. Nothing here is wired into handlers or backends;
|
||||
//! backup export, restore orchestration, and the admin API build on these
|
||||
//! types in follow-up changes.
|
||||
//! The contract side defines the versioned backup manifest, the per-backend
|
||||
//! responsibility matrix, typed failure modes, and the restore dry-run
|
||||
//! report. [`local_export`] implements the producer side for the Local
|
||||
//! backend as a crate-internal API; restore orchestration and the admin API
|
||||
//! build on these pieces in follow-up changes.
|
||||
//!
|
||||
//! # Bundle model
|
||||
//!
|
||||
@@ -50,6 +50,7 @@
|
||||
mod capability;
|
||||
mod dry_run;
|
||||
mod error;
|
||||
pub mod local_export;
|
||||
mod manifest;
|
||||
|
||||
pub use capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
|
||||
@@ -57,6 +58,10 @@ pub use dry_run::{
|
||||
ExternalDependencyMismatch, RestoreBlocker, RestoreBlockerCode, RestoreConflict, RestoreConflictKind, RestoreDryRunReport,
|
||||
};
|
||||
pub use error::BackupError;
|
||||
pub use local_export::{
|
||||
BackupKek, LOCAL_BUNDLE_MANIFEST_FILE, LocalBackupExportRequest, decrypt_bundle_artifact, export_local_backup,
|
||||
read_local_bundle_manifest,
|
||||
};
|
||||
pub use manifest::{
|
||||
AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest,
|
||||
DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation, ReservedSlot,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
+22
-508
@@ -27,12 +27,6 @@
|
||||
//! retried automatically: a response lost after the server applied the write
|
||||
//! would otherwise be replayed into duplicate side effects (extra key versions,
|
||||
//! repeated deletes).
|
||||
//!
|
||||
//! Every execution also records operation metrics (attempt failures by retry
|
||||
//! class, terminal outcome, attempts used, wall-clock duration) through the
|
||||
//! process-global `metrics` recorder. Metric labels carry only static enum
|
||||
//! values — operation names, classes, outcomes — never key identifiers, key
|
||||
//! material, ciphertext, or tokens.
|
||||
|
||||
use std::future::Future;
|
||||
use std::time::Duration;
|
||||
@@ -206,130 +200,6 @@ fn equal_jitter(rng: &mut impl RngExt, cap: Duration) -> Duration {
|
||||
half + Duration::from_nanos(rng.random_range(0..=spread))
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Metrics
|
||||
//
|
||||
// Every execution is recorded here, at the single choke point all backend
|
||||
// calls flow through, so instrumenting a new call site costs nothing beyond
|
||||
// naming its operation. Label values are exclusively static enum strings
|
||||
// (operation names, classes, outcomes) — key identifiers, key material,
|
||||
// ciphertext, and tokens must never reach a metric label.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Counter: operations executed, by `operation`, `op_class`, and `outcome`.
|
||||
const METRIC_OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
|
||||
/// Counter: failed attempts, by `operation` and `error_class` (including
|
||||
/// `attempt_timeout` for attempts cut off by the per-attempt timeout).
|
||||
const METRIC_ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
|
||||
/// Histogram: wall-clock duration of a whole operation (attempts plus
|
||||
/// backoff), in seconds, by `operation` and `outcome`.
|
||||
const METRIC_OPERATION_DURATION_SECONDS: &str = "rustfs_kms_backend_operation_duration_seconds";
|
||||
/// Histogram: attempts one operation used before completing, by `operation`
|
||||
/// and `outcome`.
|
||||
const METRIC_OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
|
||||
|
||||
impl OpClass {
|
||||
fn as_label(self) -> &'static str {
|
||||
match self {
|
||||
OpClass::ReadIdempotent => "read_idempotent",
|
||||
OpClass::MutatingNonIdempotent => "mutating_non_idempotent",
|
||||
OpClass::Auth => "auth",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ErrorClass {
|
||||
fn as_label(self) -> &'static str {
|
||||
match self {
|
||||
ErrorClass::RetryableConn => "retryable_conn",
|
||||
ErrorClass::RetryableStatus => "retryable_status",
|
||||
ErrorClass::Fatal => "fatal",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// How one policy execution terminated, for the `outcome` metric label.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum Outcome {
|
||||
Success,
|
||||
/// A fatal-classified failure ended the operation on its first observation.
|
||||
Fatal,
|
||||
/// The attempt budget ran out; the last failure was retryable (including a
|
||||
/// timed-out final attempt).
|
||||
BudgetExhausted,
|
||||
/// The operation deadline ran out before another attempt could complete.
|
||||
DeadlineExceeded,
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
impl Outcome {
|
||||
fn as_label(self) -> &'static str {
|
||||
match self {
|
||||
Outcome::Success => "success",
|
||||
Outcome::Fatal => "fatal",
|
||||
Outcome::BudgetExhausted => "budget_exhausted",
|
||||
Outcome::DeadlineExceeded => "deadline_exceeded",
|
||||
Outcome::Cancelled => "cancelled",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Register metric descriptions once per process.
|
||||
fn describe_metrics() {
|
||||
static DESCRIBE: std::sync::Once = std::sync::Once::new();
|
||||
DESCRIBE.call_once(|| {
|
||||
metrics::describe_counter!(
|
||||
METRIC_OPERATIONS_TOTAL,
|
||||
"Total KMS backend operations executed under the operation policy, by operation, operation class, and outcome"
|
||||
);
|
||||
metrics::describe_counter!(
|
||||
METRIC_ATTEMPT_FAILURES_TOTAL,
|
||||
"Total failed KMS backend attempts, by operation and retry classification"
|
||||
);
|
||||
metrics::describe_histogram!(
|
||||
METRIC_OPERATION_DURATION_SECONDS,
|
||||
"Wall-clock duration of KMS backend operations including retries and backoff, in seconds"
|
||||
);
|
||||
metrics::describe_histogram!(
|
||||
METRIC_OPERATION_ATTEMPTS,
|
||||
"Number of attempts a KMS backend operation used before completing"
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/// Record one failed attempt with its retry classification.
|
||||
fn record_attempt_failure(operation: &'static str, error_class: &'static str) {
|
||||
metrics::counter!(
|
||||
METRIC_ATTEMPT_FAILURES_TOTAL,
|
||||
"operation" => operation,
|
||||
"error_class" => error_class
|
||||
)
|
||||
.increment(1);
|
||||
}
|
||||
|
||||
/// Record the terminal outcome of one policy execution.
|
||||
fn record_operation(operation: &'static str, class: OpClass, outcome: Outcome, attempts: u32, elapsed: Duration) {
|
||||
metrics::counter!(
|
||||
METRIC_OPERATIONS_TOTAL,
|
||||
"operation" => operation,
|
||||
"op_class" => class.as_label(),
|
||||
"outcome" => outcome.as_label()
|
||||
)
|
||||
.increment(1);
|
||||
metrics::histogram!(
|
||||
METRIC_OPERATION_DURATION_SECONDS,
|
||||
"operation" => operation,
|
||||
"outcome" => outcome.as_label()
|
||||
)
|
||||
.record(elapsed.as_secs_f64());
|
||||
metrics::histogram!(
|
||||
METRIC_OPERATION_ATTEMPTS,
|
||||
"operation" => operation,
|
||||
"outcome" => outcome.as_label()
|
||||
)
|
||||
.record(f64::from(attempts));
|
||||
}
|
||||
|
||||
/// Run `attempt` under the policy.
|
||||
///
|
||||
/// Each attempt is bounded by `attempt_timeout` (further capped by whatever is
|
||||
@@ -360,37 +230,13 @@ where
|
||||
/// [`execute`] with an injectable jitter source so tests can pin deterministic
|
||||
/// backoff durations instead of asserting around random sleeps.
|
||||
pub(crate) async fn execute_with_jitter<T, F, Fut, J>(
|
||||
operation: &'static str,
|
||||
class: OpClass,
|
||||
policy: &RetryPolicy,
|
||||
cancel: &CancellationToken,
|
||||
jitter: J,
|
||||
attempt: F,
|
||||
) -> Result<T>
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: Future<Output = std::result::Result<T, AttemptError>>,
|
||||
J: FnMut(Duration) -> Duration,
|
||||
{
|
||||
describe_metrics();
|
||||
let started = Instant::now();
|
||||
let mut attempts_made = 0u32;
|
||||
let (outcome, result) = drive_attempts(operation, class, policy, cancel, jitter, attempt, &mut attempts_made).await;
|
||||
record_operation(operation, class, outcome, attempts_made, started.elapsed());
|
||||
result
|
||||
}
|
||||
|
||||
/// The attempt loop behind [`execute_with_jitter`], returning the terminal
|
||||
/// outcome alongside the result so the caller can record it exactly once.
|
||||
async fn drive_attempts<T, F, Fut, J>(
|
||||
operation: &'static str,
|
||||
class: OpClass,
|
||||
policy: &RetryPolicy,
|
||||
cancel: &CancellationToken,
|
||||
mut jitter: J,
|
||||
mut attempt: F,
|
||||
attempts_made: &mut u32,
|
||||
) -> (Outcome, Result<T>)
|
||||
) -> Result<T>
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: Future<Output = std::result::Result<T, AttemptError>>,
|
||||
@@ -401,68 +247,48 @@ where
|
||||
|
||||
let mut attempt_no = 0u32;
|
||||
loop {
|
||||
attempt_no += 1;
|
||||
if cancel.is_cancelled() {
|
||||
return (
|
||||
Outcome::Cancelled,
|
||||
Err(KmsError::operation_cancelled(format!(
|
||||
"{operation} cancelled before attempt {}",
|
||||
attempt_no + 1
|
||||
))),
|
||||
);
|
||||
return Err(KmsError::operation_cancelled(format!(
|
||||
"{operation} cancelled before attempt {attempt_no}"
|
||||
)));
|
||||
}
|
||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||
if remaining.is_zero() {
|
||||
return (
|
||||
Outcome::DeadlineExceeded,
|
||||
Err(KmsError::operation_timed_out(format!(
|
||||
"{operation} exceeded operation deadline of {:?}",
|
||||
policy.op_deadline
|
||||
))),
|
||||
);
|
||||
return Err(KmsError::operation_timed_out(format!(
|
||||
"{operation} exceeded operation deadline of {:?}",
|
||||
policy.op_deadline
|
||||
)));
|
||||
}
|
||||
|
||||
attempt_no += 1;
|
||||
*attempts_made = attempt_no;
|
||||
let attempt_budget = policy.attempt_timeout.min(remaining);
|
||||
let outcome = tokio::select! {
|
||||
biased;
|
||||
_ = cancel.cancelled() => {
|
||||
return (
|
||||
Outcome::Cancelled,
|
||||
Err(KmsError::operation_cancelled(format!("{operation} cancelled during attempt {attempt_no}"))),
|
||||
);
|
||||
return Err(KmsError::operation_cancelled(format!("{operation} cancelled during attempt {attempt_no}")));
|
||||
}
|
||||
outcome = tokio::time::timeout(attempt_budget, attempt()) => outcome,
|
||||
};
|
||||
|
||||
let failure = match outcome {
|
||||
Ok(Ok(value)) => return (Outcome::Success, Ok(value)),
|
||||
Ok(Err(failure)) => {
|
||||
record_attempt_failure(operation, failure.class.as_label());
|
||||
failure
|
||||
}
|
||||
Err(_) => {
|
||||
record_attempt_failure(operation, "attempt_timeout");
|
||||
AttemptError {
|
||||
class: ErrorClass::RetryableConn,
|
||||
error: KmsError::operation_timed_out(format!(
|
||||
"{operation} attempt {attempt_no} timed out after {attempt_budget:?}"
|
||||
)),
|
||||
}
|
||||
}
|
||||
Ok(Ok(value)) => return Ok(value),
|
||||
Ok(Err(failure)) => failure,
|
||||
Err(_) => AttemptError {
|
||||
class: ErrorClass::RetryableConn,
|
||||
error: KmsError::operation_timed_out(format!(
|
||||
"{operation} attempt {attempt_no} timed out after {attempt_budget:?}"
|
||||
)),
|
||||
},
|
||||
};
|
||||
|
||||
if failure.class == ErrorClass::Fatal {
|
||||
return (Outcome::Fatal, Err(failure.error));
|
||||
}
|
||||
if attempt_no >= max_attempts {
|
||||
return (Outcome::BudgetExhausted, Err(failure.error));
|
||||
if failure.class == ErrorClass::Fatal || attempt_no >= max_attempts {
|
||||
return Err(failure.error);
|
||||
}
|
||||
|
||||
let backoff = jitter(backoff_cap(policy, attempt_no));
|
||||
if backoff >= deadline.saturating_duration_since(Instant::now()) {
|
||||
// Not enough deadline budget left for another attempt.
|
||||
return (Outcome::DeadlineExceeded, Err(failure.error));
|
||||
return Err(failure.error);
|
||||
}
|
||||
tracing::warn!(
|
||||
operation,
|
||||
@@ -474,10 +300,7 @@ where
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = cancel.cancelled() => {
|
||||
return (
|
||||
Outcome::Cancelled,
|
||||
Err(KmsError::operation_cancelled(format!("{operation} cancelled during retry backoff"))),
|
||||
);
|
||||
return Err(KmsError::operation_cancelled(format!("{operation} cancelled during retry backoff")));
|
||||
}
|
||||
_ = tokio::time::sleep(backoff) => {}
|
||||
}
|
||||
@@ -805,313 +628,4 @@ mod tests {
|
||||
assert_eq!(classify_vaultrs(&ClientError::ResponseEmptyError), ErrorClass::Fatal);
|
||||
assert_eq!(classify_vaultrs(&ClientError::InvalidLoginMethodError), ErrorClass::Fatal);
|
||||
}
|
||||
|
||||
// -- Metric emission ----------------------------------------------------
|
||||
//
|
||||
// Each test installs a thread-local debugging recorder and drives a
|
||||
// paused-clock current-thread runtime inside it, so the emitted metrics
|
||||
// (including virtual-clock durations) are fully deterministic.
|
||||
|
||||
use metrics_util::MetricKind;
|
||||
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
|
||||
|
||||
type MetricEntry = (
|
||||
metrics_util::CompositeKey,
|
||||
Option<metrics::Unit>,
|
||||
Option<metrics::SharedString>,
|
||||
DebugValue,
|
||||
);
|
||||
|
||||
/// Run `test` on a paused current-thread runtime under a debugging
|
||||
/// recorder and return one snapshot of everything it emitted.
|
||||
///
|
||||
/// A single snapshot per test on purpose: `Snapshotter::snapshot` drains
|
||||
/// the recorded state, so taking it per assertion would only show the
|
||||
/// first assertion any data.
|
||||
fn record_metrics<Out>(test: impl FnOnce() -> std::pin::Pin<Box<dyn Future<Output = Out>>>) -> (Vec<MetricEntry>, Out) {
|
||||
let recorder = DebuggingRecorder::new();
|
||||
let snapshotter = recorder.snapshotter();
|
||||
let out = metrics::with_local_recorder(&recorder, || {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_time()
|
||||
.start_paused(true)
|
||||
.build()
|
||||
.expect("current-thread runtime must build");
|
||||
runtime.block_on(test())
|
||||
});
|
||||
(snapshotter.snapshot().into_vec(), out)
|
||||
}
|
||||
|
||||
fn labels_match(key: &metrics::Key, labels: &[(&str, &str)]) -> bool {
|
||||
labels
|
||||
.iter()
|
||||
.all(|(label, expected)| key.labels().any(|l| l.key() == *label && l.value() == *expected))
|
||||
}
|
||||
|
||||
fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> u64 {
|
||||
snapshot
|
||||
.iter()
|
||||
.filter_map(|(composite, _unit, _description, value)| {
|
||||
let matches = composite.kind() == MetricKind::Counter
|
||||
&& composite.key().name() == name
|
||||
&& labels_match(composite.key(), labels);
|
||||
match (matches, value) {
|
||||
(true, DebugValue::Counter(count)) => Some(*count),
|
||||
_ => None,
|
||||
}
|
||||
})
|
||||
.sum()
|
||||
}
|
||||
|
||||
fn histogram_values(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> Vec<f64> {
|
||||
snapshot
|
||||
.iter()
|
||||
.filter_map(|(composite, _unit, _description, value)| {
|
||||
let matches = composite.kind() == MetricKind::Histogram
|
||||
&& composite.key().name() == name
|
||||
&& labels_match(composite.key(), labels);
|
||||
match (matches, value) {
|
||||
(true, DebugValue::Histogram(values)) => Some(values),
|
||||
_ => None,
|
||||
}
|
||||
})
|
||||
.flatten()
|
||||
.map(|value| value.into_inner())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metrics_record_retried_success_with_attempts_and_duration() {
|
||||
let calls_in_test = Arc::new(AtomicU32::new(0));
|
||||
let (snapshot, ()) = record_metrics(move || {
|
||||
Box::pin(async move {
|
||||
let policy = policy_of(1_000, 60_000, 3, 100, 2_000);
|
||||
let cancel = CancellationToken::new();
|
||||
let calls_in_attempt = calls_in_test.clone();
|
||||
execute_with_jitter("metrics_read", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, move || {
|
||||
let calls = calls_in_attempt.clone();
|
||||
async move {
|
||||
if calls.fetch_add(1, Ordering::SeqCst) < 2 {
|
||||
Err(AttemptError {
|
||||
class: ErrorClass::RetryableStatus,
|
||||
error: KmsError::backend_error("throttled (429)"),
|
||||
})
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("retries within budget must succeed");
|
||||
})
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
METRIC_OPERATIONS_TOTAL,
|
||||
&[
|
||||
("operation", "metrics_read"),
|
||||
("op_class", "read_idempotent"),
|
||||
("outcome", "success")
|
||||
]
|
||||
),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
METRIC_ATTEMPT_FAILURES_TOTAL,
|
||||
&[("operation", "metrics_read"), ("error_class", "retryable_status")]
|
||||
),
|
||||
2
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_values(
|
||||
&snapshot,
|
||||
METRIC_OPERATION_ATTEMPTS,
|
||||
&[("operation", "metrics_read"), ("outcome", "success")]
|
||||
),
|
||||
vec![3.0]
|
||||
);
|
||||
// Full-cap backoffs of 100ms and 200ms on the paused clock.
|
||||
let durations = histogram_values(
|
||||
&snapshot,
|
||||
METRIC_OPERATION_DURATION_SECONDS,
|
||||
&[("operation", "metrics_read"), ("outcome", "success")],
|
||||
);
|
||||
assert_eq!(durations.len(), 1);
|
||||
assert!((durations[0] - 0.3).abs() < 1e-9, "expected 0.3s of virtual backoff, got {durations:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metrics_record_fatal_outcome_with_single_attempt() {
|
||||
let (snapshot, ()) = record_metrics(|| {
|
||||
Box::pin(async {
|
||||
let policy = policy_of(1_000, 60_000, 5, 100, 2_000);
|
||||
let cancel = CancellationToken::new();
|
||||
let result: Result<()> =
|
||||
execute_with_jitter("metrics_fatal", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || async {
|
||||
Err(AttemptError {
|
||||
class: ErrorClass::Fatal,
|
||||
error: KmsError::access_denied("permission denied (403)"),
|
||||
})
|
||||
})
|
||||
.await;
|
||||
result.expect_err("a fatal failure must end the operation");
|
||||
})
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
METRIC_OPERATIONS_TOTAL,
|
||||
&[("operation", "metrics_fatal"), ("outcome", "fatal")]
|
||||
),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
METRIC_ATTEMPT_FAILURES_TOTAL,
|
||||
&[("operation", "metrics_fatal"), ("error_class", "fatal")]
|
||||
),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_values(
|
||||
&snapshot,
|
||||
METRIC_OPERATION_ATTEMPTS,
|
||||
&[("operation", "metrics_fatal"), ("outcome", "fatal")]
|
||||
),
|
||||
vec![1.0]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metrics_record_mutating_budget_exhausted_after_one_attempt() {
|
||||
let (snapshot, ()) = record_metrics(|| {
|
||||
Box::pin(async {
|
||||
let policy = policy_of(1_000, 60_000, 5, 100, 2_000);
|
||||
let cancel = CancellationToken::new();
|
||||
let result: Result<()> = execute_with_jitter(
|
||||
"metrics_rotate",
|
||||
OpClass::MutatingNonIdempotent,
|
||||
&policy,
|
||||
&cancel,
|
||||
full_jitter,
|
||||
|| async { Err(retryable_conn_error()) },
|
||||
)
|
||||
.await;
|
||||
result.expect_err("a mutating operation must not retry a retryable failure");
|
||||
})
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
METRIC_OPERATIONS_TOTAL,
|
||||
&[
|
||||
("operation", "metrics_rotate"),
|
||||
("op_class", "mutating_non_idempotent"),
|
||||
("outcome", "budget_exhausted")
|
||||
]
|
||||
),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_values(
|
||||
&snapshot,
|
||||
METRIC_OPERATION_ATTEMPTS,
|
||||
&[("operation", "metrics_rotate"), ("outcome", "budget_exhausted")]
|
||||
),
|
||||
vec![1.0]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metrics_record_timeouts_and_deadline_outcome() {
|
||||
let (snapshot, ()) = record_metrics(|| {
|
||||
Box::pin(async {
|
||||
// Hung attempts: 10s each against a 25s deadline (see
|
||||
// total_duration_never_exceeds_deadline for the timeline).
|
||||
let policy = policy_of(10_000, 25_000, 5, 100, 2_000);
|
||||
let cancel = CancellationToken::new();
|
||||
let result: Result<()> =
|
||||
execute_with_jitter("metrics_hung", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || {
|
||||
std::future::pending::<AttemptResult<()>>()
|
||||
})
|
||||
.await;
|
||||
result.expect_err("hung attempts must exhaust the deadline");
|
||||
})
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
METRIC_OPERATIONS_TOTAL,
|
||||
&[("operation", "metrics_hung"), ("outcome", "deadline_exceeded")]
|
||||
),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
METRIC_ATTEMPT_FAILURES_TOTAL,
|
||||
&[("operation", "metrics_hung"), ("error_class", "attempt_timeout")]
|
||||
),
|
||||
3
|
||||
);
|
||||
let durations = histogram_values(
|
||||
&snapshot,
|
||||
METRIC_OPERATION_DURATION_SECONDS,
|
||||
&[("operation", "metrics_hung"), ("outcome", "deadline_exceeded")],
|
||||
);
|
||||
assert_eq!(durations.len(), 1);
|
||||
assert!((durations[0] - 25.0).abs() < 1e-9, "expected the full 25s deadline, got {durations:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metrics_record_cancelled_outcome() {
|
||||
let calls_in_test = Arc::new(AtomicU32::new(0));
|
||||
let (snapshot, ()) = record_metrics(move || {
|
||||
Box::pin(async move {
|
||||
let policy = policy_of(1_000, 600_000, 5, 10_000, 10_000);
|
||||
let cancel = CancellationToken::new();
|
||||
let canceller = cancel.clone();
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
canceller.cancel();
|
||||
});
|
||||
let calls_in_attempt = calls_in_test.clone();
|
||||
let result: Result<()> =
|
||||
execute_with_jitter("metrics_cancel", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, move || {
|
||||
let calls = calls_in_attempt.clone();
|
||||
async move {
|
||||
calls.fetch_add(1, Ordering::SeqCst);
|
||||
Err(retryable_conn_error())
|
||||
}
|
||||
})
|
||||
.await;
|
||||
result.expect_err("cancellation must abort the backoff");
|
||||
})
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
METRIC_OPERATIONS_TOTAL,
|
||||
&[("operation", "metrics_cancel"), ("outcome", "cancelled")]
|
||||
),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
histogram_values(
|
||||
&snapshot,
|
||||
METRIC_OPERATION_ATTEMPTS,
|
||||
&[("operation", "metrics_cancel"), ("outcome", "cancelled")]
|
||||
),
|
||||
vec![1.0]
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,255 +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.
|
||||
|
||||
//! Fault-injection matrix for the Vault backend operation policy.
|
||||
//!
|
||||
//! Offline cases run against locally injected transport faults (a closed
|
||||
//! port, a listener that never responds) — deterministic, no external
|
||||
//! dependencies. Real-Vault cases are `#[ignore]`d and need a dev Vault
|
||||
//! (default `http://127.0.0.1:8200`, override with `RUSTFS_KMS_VAULT_ADDR`).
|
||||
//!
|
||||
//! Throttling (429) and recoverable 5xx responses cannot be forced on a stock
|
||||
//! dev Vault, so their retry and metric behavior is pinned deterministically
|
||||
//! by the scripted-Vault wiring tests in `backends::vault` and the engine
|
||||
//! tests in `policy.rs`. Pointing `RUSTFS_KMS_VAULT_ADDR` at a
|
||||
//! fault-injecting proxy reuses the ignored cases here unchanged.
|
||||
//!
|
||||
//! Every case installs a thread-local debugging metrics recorder and drives a
|
||||
//! current-thread runtime inside it, so the policy metrics double as the
|
||||
//! request-count assertion even against a real server.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use metrics_util::MetricKind;
|
||||
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
|
||||
use rustfs_kms::backends::KmsClient;
|
||||
use rustfs_kms::backends::vault::VaultKmsClient;
|
||||
use rustfs_kms::{KmsConfig, KmsError, VaultAuthMethod, VaultConfig};
|
||||
|
||||
const OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
|
||||
const ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
|
||||
|
||||
fn vault_config(address: &str, token: &str) -> VaultConfig {
|
||||
VaultConfig {
|
||||
address: address.to_string(),
|
||||
auth_method: VaultAuthMethod::Token {
|
||||
token: token.to_string(),
|
||||
},
|
||||
namespace: None,
|
||||
mount_path: "transit".to_string(),
|
||||
kv_mount: "secret".to_string(),
|
||||
key_path_prefix: "rustfs/kms/fault-injection".to_string(),
|
||||
tls: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn kms_config(attempt_timeout: Duration, retry_attempts: u32) -> KmsConfig {
|
||||
KmsConfig {
|
||||
timeout: attempt_timeout,
|
||||
retry_attempts,
|
||||
..KmsConfig::default()
|
||||
}
|
||||
}
|
||||
|
||||
type MetricEntry = (
|
||||
metrics_util::CompositeKey,
|
||||
Option<metrics::Unit>,
|
||||
Option<metrics::SharedString>,
|
||||
DebugValue,
|
||||
);
|
||||
|
||||
/// Run `test` on a current-thread runtime under a debugging metrics recorder
|
||||
/// and return one snapshot of everything it emitted.
|
||||
///
|
||||
/// A single snapshot per test on purpose: `Snapshotter::snapshot` drains the
|
||||
/// recorded state, so taking it per assertion would only show the first
|
||||
/// assertion any data.
|
||||
fn record_metrics(test: impl FnOnce() -> std::pin::Pin<Box<dyn std::future::Future<Output = ()>>>) -> Vec<MetricEntry> {
|
||||
let recorder = DebuggingRecorder::new();
|
||||
let snapshotter = recorder.snapshotter();
|
||||
metrics::with_local_recorder(&recorder, || {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.expect("current-thread runtime must build");
|
||||
runtime.block_on(test());
|
||||
});
|
||||
snapshotter.snapshot().into_vec()
|
||||
}
|
||||
|
||||
/// Sum of counters with `name` whose labels include every `(key, value)` pair.
|
||||
fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> u64 {
|
||||
snapshot
|
||||
.iter()
|
||||
.filter_map(|(composite, _unit, _description, value)| {
|
||||
let key = composite.key();
|
||||
let matches = composite.kind() == MetricKind::Counter
|
||||
&& key.name() == name
|
||||
&& labels
|
||||
.iter()
|
||||
.all(|(label, expected)| key.labels().any(|l| l.key() == *label && l.value() == *expected));
|
||||
match (matches, value) {
|
||||
(true, DebugValue::Counter(count)) => Some(*count),
|
||||
_ => None,
|
||||
}
|
||||
})
|
||||
.sum()
|
||||
}
|
||||
|
||||
/// Connection refused: connection-class failures are retried up to the
|
||||
/// configured budget, then surface as a backend error.
|
||||
#[test]
|
||||
fn connection_refused_is_retried_within_budget() {
|
||||
// Reserve a loopback port and release it so nothing is listening there.
|
||||
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a loopback port");
|
||||
let address = format!("http://{}", listener.local_addr().expect("reserved port addr"));
|
||||
drop(listener);
|
||||
|
||||
let snapshot = record_metrics(|| {
|
||||
Box::pin(async move {
|
||||
let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_secs(2), 2))
|
||||
.await
|
||||
.expect("client construction performs no network calls");
|
||||
let error = client
|
||||
.describe_key("fault-injection-refused", None)
|
||||
.await
|
||||
.expect_err("a refused connection must fail the operation");
|
||||
assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
|
||||
})
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]),
|
||||
2,
|
||||
"both budgeted attempts must observe the refused connection"
|
||||
);
|
||||
assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1);
|
||||
// The static-token login records its own success; the Vault read must not.
|
||||
assert_eq!(
|
||||
counter_value(
|
||||
&snapshot,
|
||||
OPERATIONS_TOTAL,
|
||||
&[("operation", "vault_kv2_read_key"), ("outcome", "success")]
|
||||
),
|
||||
0
|
||||
);
|
||||
}
|
||||
|
||||
/// Stalled connection: a server that accepts but never responds is cut off by
|
||||
/// the per-attempt timeout (either the policy timer or the equally sized HTTP
|
||||
/// client timeout, whichever fires first) instead of hanging forever.
|
||||
#[test]
|
||||
fn stalled_connection_is_cut_off_by_the_attempt_timeout() {
|
||||
let snapshot = record_metrics(|| {
|
||||
Box::pin(async move {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("bind stall listener");
|
||||
let address = format!("http://{}", listener.local_addr().expect("stall listener addr"));
|
||||
// Accept and park every connection without ever responding.
|
||||
tokio::spawn(async move {
|
||||
let mut parked = Vec::new();
|
||||
loop {
|
||||
let Ok((socket, _)) = listener.accept().await else { return };
|
||||
parked.push(socket);
|
||||
}
|
||||
});
|
||||
|
||||
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 = client
|
||||
.describe_key("fault-injection-stalled", None)
|
||||
.await
|
||||
.expect_err("a stalled request must be cut off by the attempt timeout");
|
||||
assert!(
|
||||
matches!(error, KmsError::OperationTimedOut { .. } | KmsError::BackendError { .. }),
|
||||
"got {error:?}"
|
||||
);
|
||||
})
|
||||
});
|
||||
|
||||
// The policy timer reports attempt_timeout; the client-level HTTP timeout
|
||||
// surfaces as a connection-class failure. Either way it is exactly one
|
||||
// attempt that was cut off.
|
||||
let cut_off = counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "attempt_timeout")])
|
||||
+ counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]);
|
||||
assert_eq!(cut_off, 1, "the single budgeted attempt must be cut off by a timeout");
|
||||
assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1);
|
||||
}
|
||||
|
||||
fn real_vault_address() -> String {
|
||||
std::env::var("RUSTFS_KMS_VAULT_ADDR").unwrap_or_else(|_| "http://127.0.0.1:8200".to_string())
|
||||
}
|
||||
|
||||
/// Invalid token against a real Vault: the 403 is fatal — exactly one
|
||||
/// attempt, no retry, and the operation fails closed.
|
||||
#[test]
|
||||
#[ignore] // Requires a running Vault dev server
|
||||
fn real_vault_invalid_token_is_fatal_and_never_retried() {
|
||||
let snapshot = record_metrics(|| {
|
||||
Box::pin(async {
|
||||
let config = vault_config(&real_vault_address(), "fault-injection-invalid-token");
|
||||
let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
|
||||
.await
|
||||
.expect("client construction performs no network calls");
|
||||
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:?}");
|
||||
})
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "fatal")]),
|
||||
1,
|
||||
"a 403 must be observed by exactly one attempt"
|
||||
);
|
||||
assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "fatal")]), 1);
|
||||
assert_eq!(
|
||||
counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_status")]),
|
||||
0,
|
||||
"an auth failure must never be classified as retryable"
|
||||
);
|
||||
}
|
||||
|
||||
/// Healthy read against a real Vault: a missing key resolves in one attempt
|
||||
/// (404 is fatal for retry purposes) and records a fatal outcome rather than
|
||||
/// burning the retry budget.
|
||||
#[test]
|
||||
#[ignore] // Requires a running Vault dev server
|
||||
fn real_vault_missing_key_is_resolved_in_one_attempt() {
|
||||
let token = std::env::var("RUSTFS_KMS_VAULT_TOKEN").unwrap_or_else(|_| "dev-only-token".to_string());
|
||||
let snapshot = record_metrics(|| {
|
||||
Box::pin(async move {
|
||||
let config = vault_config(&real_vault_address(), &token);
|
||||
let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
|
||||
.await
|
||||
.expect("client construction performs no network calls");
|
||||
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:?}");
|
||||
})
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "fatal")]),
|
||||
1,
|
||||
"a 404 must be observed by exactly one attempt"
|
||||
);
|
||||
assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "fatal")]), 1);
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -28,30 +28,6 @@ documentation = "https://docs.rs/rustfs-policy/latest/rustfs_policy/"
|
||||
[lints]
|
||||
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",
|
||||
]
|
||||
|
||||
[dependencies]
|
||||
rustfs-credentials = { workspace = true }
|
||||
rustfs-config = { workspace = true, features = ["opa"] }
|
||||
@@ -73,7 +49,6 @@ moka = { workspace = true, features = ["future"] }
|
||||
async-trait.workspace = true
|
||||
futures.workspace = true
|
||||
pollster.workspace = true
|
||||
hotpath.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
pollster.workspace = true
|
||||
|
||||
@@ -32,15 +32,10 @@ impl Args {
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AuthZPlugin {
|
||||
client: OpaHttpClient,
|
||||
client: reqwest::Client,
|
||||
args: Args,
|
||||
}
|
||||
|
||||
#[cfg(feature = "hotpath")]
|
||||
type OpaHttpClient = hotpath::wrap::reqwest::Client;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
type OpaHttpClient = reqwest::Client;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum OpaConfigError {
|
||||
#[error("Missing required env var: {0}")]
|
||||
@@ -146,9 +141,6 @@ impl AuthZPlugin {
|
||||
reqwest::Client::new()
|
||||
});
|
||||
|
||||
#[cfg(feature = "hotpath")]
|
||||
let client = hotpath::http!(client, label = "Policy::OPA");
|
||||
|
||||
Self { client, args: config }
|
||||
}
|
||||
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
+2
-26
@@ -30,32 +30,8 @@ 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"]
|
||||
|
||||
[dependencies]
|
||||
hotpath.workspace = true
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -2531,7 +2531,6 @@ where
|
||||
}
|
||||
|
||||
#[instrument(skip_all)]
|
||||
#[hotpath::measure]
|
||||
async fn run_data_scanner_cycle(
|
||||
ctx: &CancellationToken,
|
||||
storeapi: &Arc<ECStore>,
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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>,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -60,137 +60,15 @@ 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]
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user