Merge branch 'main' into overtrue/kms-1638-d2-minio-sse-read

chore: adjudicate 26 bare dead_code allows across five crates (#6187 )
Remove every bare `#[allow(dead_code)]` in io-core, object-capacity, targets, rio, and scanner. Each allow was stripped first and clippy was then asked which ones the compiler actually missed, so the verdicts rest on the diagnostic rather than on inspection. 23 were inert: they sat on `pub fn`s inside `pub mod`s, where `dead_code` does not apply, or on scanner integration-test helpers that the tests in the same file do call. The remaining 3 are in rio's private `compress_index` module and the code behind them is deleted rather than annotated. `remove_index_headers` is dead and also wrong — after skipping the 4-byte chunk header it matches against `S2_INDEX_TRAILER` where `S2_INDEX_HEADER` sits, so it returns `None` for every well-formed index; rio-v2 carries the correct equivalent that is actually in use. `restore_index_headers` is its unreachable counterpart, likewise duplicated live in rio-v2. `Index::reset` is a private method with no caller. Refs backlog#1823
2026-08-18 10:43:15 +00:00 · 2026-08-18 12:46:12 +08:00 · 2026-08-18 12:45:42 +08:00 · 2026-08-18 12:43:27 +08:00 · 2026-08-18 04:35:36 +00:00 · 2026-08-18 12:21:49 +08:00
36 changed files with 1862 additions and 542 deletions
@@ -964,9 +964,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-kms"
-version = "1.114.0"
+version = "1.115.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c0b7d906608ee41e7ddea9983577ba82200435644d567d63dc34e822e088b453"
+checksum = "d5b034f8b7ceadb873d0bc607c30bb4b0be68e09a84c837174e7c2c6878ff882"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -990,9 +990,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-s3"
-version = "1.141.0"
+version = "1.142.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9f9420d3a2467eed22ed3635ca653653162c386a0b0f65c78189f9bd3c1379e"
+checksum = "f9e15a5c55e05f4b0b7e483160b3c85cccdf77cff02c95504f3e71d460855cd2"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1027,9 +1027,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-sso"
-version = "1.105.0"
+version = "1.106.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6ffd0fbe7873cb548a7aa60f9573c268fff94155397fd4f14dc9f1ecaaab8516"
+checksum = "2d0efcee834347b6705eca3eea2defd88242f43774f55d7326604222e3c86260"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1053,9 +1053,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-ssooidc"
-version = "1.107.0"
+version = "1.108.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "175763eb222a46377df7aa257a3bca980ab3e96703fefc8f4d0b8da6ad2e254c"
+checksum = "a59312a04cf19c962cfee32b64ecfee758f8786407ff6da5b30fff46ae96f201"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1079,9 +1079,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-sts"
-version = "1.110.0"
+version = "1.111.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dd8b14781dfbff48984017d57167b6ea0b6471c6920ec52b44a2677c7feb3c13"
+checksum = "120e7eb63457a9e547f9986fe3b273f77c43679da4d04f46359fa881c5e19b6e"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1598,7 +1598,7 @@ version = "0.10.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
 dependencies = [
- "generic-array 0.14.9",
+ "generic-array 0.14.7",
 ]

 [[package]]
@@ -1617,7 +1617,7 @@ version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93"
 dependencies = [
- "generic-array 0.14.9",
+ "generic-array 0.14.7",
 ]

 [[package]]
@@ -1858,9 +1858,9 @@ dependencies = [

 [[package]]
 name = "cc"
-version = "1.4.2"
+version = "1.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d262e149917187838d5b42777c8253bcb64500067342904e7d429499a6f277e"
+checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
 dependencies = [
 "find-msvc-tools",
 "jobserver",
@@ -1968,7 +1968,7 @@ version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
 dependencies = [
- "crypto-common 0.1.6",
+ "crypto-common 0.1.7",
 "inout 0.1.4",
 ]

@@ -2428,7 +2428,7 @@ version = "0.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76"
 dependencies = [
- "generic-array 0.14.9",
+ "generic-array 0.14.7",
 "rand_core 0.6.4",
 "subtle",
 "zeroize",
@@ -2453,11 +2453,11 @@ dependencies = [

 [[package]]
 name = "crypto-common"
-version = "0.1.6"
+version = "0.1.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
+checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
 dependencies = [
- "generic-array 0.14.9",
+ "generic-array 0.14.7",
 "typenum",
 ]

@@ -3664,7 +3664,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
 "block-buffer 0.10.4",
 "const-oid 0.9.6",
- "crypto-common 0.1.6",
+ "crypto-common 0.1.7",
 "subtle",
 ]

@@ -3924,7 +3924,7 @@ dependencies = [
 "crypto-bigint 0.5.5",
 "digest 0.10.7",
 "ff 0.13.1",
- "generic-array 0.14.9",
+ "generic-array 0.14.7",
 "group 0.13.0",
 "hkdf 0.12.4",
 "pem-rfc7468 0.7.0",
@@ -4148,9 +4148,9 @@ dependencies = [

 [[package]]
 name = "find-msvc-tools"
-version = "0.1.10"
+version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de"
+checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890"

 [[package]]
 name = "findshlibs"
@@ -4369,9 +4369,9 @@ dependencies = [

 [[package]]
 name = "generic-array"
-version = "0.14.9"
+version = "0.14.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2"
+checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
 dependencies = [
 "typenum",
 "version_check",
@@ -4380,11 +4380,11 @@ dependencies = [

 [[package]]
 name = "generic-array"
-version = "1.4.4"
+version = "1.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ab4e5aa225bc56696909483320f0ff9b600f1a971b52e07a17d70f3d9b43254b"
+checksum = "337d46834ee672ab3e48caca2cb0c78cc174fb12b3a68d0d88f99a0519a5e36e"
 dependencies = [
- "generic-array 0.14.9",
+ "generic-array 0.14.7",
 "rustversion",
 "typenum",
 ]
@@ -4726,9 +4726,9 @@ dependencies = [

 [[package]]
 name = "h2"
-version = "0.4.15"
+version = "0.4.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155"
+checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27"
 dependencies = [
 "atomic-waker",
 "bytes",
@@ -5028,9 +5028,9 @@ dependencies = [

 [[package]]
 name = "hotpath"
-version = "0.23.2"
+version = "0.23.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "62e810bedda5a467ef5c9b5c8a20763fefebc89b63ef36f7ee44a143085204a2"
+checksum = "dce755d457a63bdd0c95e4c91511daad1b58b33209543b7f38027b676f387e5e"
 dependencies = [
 "arc-swap",
 "async-channel",
@@ -5062,9 +5062,9 @@ dependencies = [

 [[package]]
 name = "hotpath-macros"
-version = "0.23.2"
+version = "0.23.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "01bdc59bfc1a9984bee2ff5da63b2f6fccbaa57cd9a4119d709524632bddf341"
+checksum = "a903af89a8429cb07790c3818bc15270b394f80af1bc254e5ccf9c7de2961770"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5073,15 +5073,15 @@ dependencies = [

 [[package]]
 name = "hotpath-macros-meta"
-version = "0.23.2"
+version = "0.23.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9216e8a01abe1e1671c376dc8736fb1bf772d7a889538d25f9e1200120ced38"
+checksum = "bcc0ab94ffbb2ee77f4a897df02b5a137a10cf24d69bda936e59aff4dd456e61"

 [[package]]
 name = "hotpath-meta"
-version = "0.23.2"
+version = "0.23.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f22a9d20435fb79511b19dae37b3607224cd98f342a410702d84657cc38fc72f"
+checksum = "053481f6cec8f775a3276c7f6e2f21123111d28261e4edc15ea7421c445964bb"
 dependencies = [
 "hotpath-macros-meta",
 ]
@@ -5280,9 +5280,9 @@ dependencies = [

 [[package]]
 name = "icu_collections"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2984d1cd16c883d7935b9e07e44071dca8d917fd52ecc02c04d5fa0b5a3f191c"
+checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513"
 dependencies = [
 "displaydoc",
 "potential_utf",
@@ -5294,9 +5294,9 @@ dependencies = [

 [[package]]
 name = "icu_locale_core"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92219b62b3e2b4d88ac5119f8904c10f8f61bf7e95b640d25ba3075e6cac2c29"
+checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb"
 dependencies = [
 "displaydoc",
 "litemap",
@@ -5307,9 +5307,9 @@ dependencies = [

 [[package]]
 name = "icu_normalizer"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c56e5ee99d6e3d33bd91c5d85458b6005a22140021cc324cea84dd0e72cff3b4"
+checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f"
 dependencies = [
 "icu_collections",
 "icu_normalizer_data",
@@ -5321,16 +5321,17 @@ dependencies = [

 [[package]]
 name = "icu_normalizer_data"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "da3be0ae77ea334f4da67c12f149704f19f81d1adf7c51cf482943e84a2bad38"
+checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0"

 [[package]]
 name = "icu_properties"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bee3b67d0ea5c2cca5003417989af8996f8604e34fb9ddf96208a033901e70de"
+checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148"
 dependencies = [
+ "displaydoc",
 "icu_collections",
 "icu_locale_core",
 "icu_properties_data",
@@ -5341,15 +5342,15 @@ dependencies = [

 [[package]]
 name = "icu_properties_data"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e2bbb201e0c04f7b4b3e14382af113e17ba4f63e2c9d2ee626b720cbce54a14"
+checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa"

 [[package]]
 name = "icu_provider"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "139c4cf31c8b5f33d7e199446eff9c1e02decfc2f0eec2c8d71f65befa45b421"
+checksum = "92a7ed671a6aad807a8651a2e1782a6598fda9ce5185dd8158549e95a91c6428"
 dependencies = [
 "displaydoc",
 "icu_locale_core",
@@ -5417,7 +5418,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01"
 dependencies = [
 "block-padding 0.3.3",
- "generic-array 0.14.9",
+ "generic-array 0.14.7",
 ]

 [[package]]
@@ -5968,9 +5969,9 @@ dependencies = [

 [[package]]
 name = "libredox"
-version = "0.1.19"
+version = "0.1.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2026a5056764a10b2bf5d56488cba40da507f5493a6a429340e2004d9ed085fa"
+checksum = "28d0a00925a9f930d679b6789b721e3a7f9ed110f41b86d2497caa780c3a070a"
 dependencies = [
 "libc",
 ]
@@ -6033,9 +6034,9 @@ checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"

 [[package]]
 name = "litemap"
-version = "0.8.2"
+version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
+checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae"

 [[package]]
 name = "local-ip-address"
@@ -6482,9 +6483,9 @@ dependencies = [

 [[package]]
 name = "mqttbytes-core-next"
-version = "0.33.3"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3ff7ae19c74aba9e0ed6e4071cd52aa364e020076fa3cc6ef17e43662f756f3c"
+checksum = "366b6ba2b4209ca4bc5ac731ccddf570d09831981eed07e5fbd63564cf0cf1aa"
 dependencies = [
 "bytes",
 "thiserror 2.0.20",
@@ -6885,7 +6886,7 @@ version = "5.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d"
 dependencies = [
- "base64 0.21.7",
+ "base64 0.22.1",
 "chrono",
 "getrandom 0.2.17",
 "http 1.5.0",
@@ -7344,9 +7345,9 @@ dependencies = [

 [[package]]
 name = "pageant"
-version = "0.2.1"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4f3a5ae18f65a85c67a77d18d42d3606c07948e3c17c1e5f74852b26589e88a5"
+checksum = "3adadc44070da6f464b0918655a12f5792c156e088d8c4082d13e27d94c3e791"
 dependencies = [
 "base16ct 1.0.0",
 "byteorder",
@@ -7728,9 +7729,9 @@ dependencies = [

 [[package]]
 name = "pkg-config"
-version = "0.3.33"
+version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e"
+checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548"

 [[package]]
 name = "plotters"
@@ -7836,9 +7837,9 @@ dependencies = [

 [[package]]
 name = "potential_utf"
-version = "0.1.5"
+version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0103b1cef7ec0cf76490e969665504990193874ea05c85ff9bab8b911d0a0564"
+checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661"
 dependencies = [
 "zerovec",
 ]
@@ -8046,7 +8047,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
 dependencies = [
 "heck 0.5.0",
- "itertools 0.10.5",
+ "itertools 0.14.0",
 "log",
 "multimap",
 "once_cell",
@@ -8066,7 +8067,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042"
 dependencies = [
 "heck 0.5.0",
- "itertools 0.10.5",
+ "itertools 0.14.0",
 "log",
 "multimap",
 "petgraph 0.8.3",
@@ -8087,7 +8088,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
 dependencies = [
 "anyhow",
- "itertools 0.10.5",
+ "itertools 0.14.0",
 "proc-macro2",
 "quote",
 "syn 2.0.119",
@@ -8100,7 +8101,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
 dependencies = [
 "anyhow",
- "itertools 0.10.5",
+ "itertools 0.14.0",
 "proc-macro2",
 "quote",
 "syn 2.0.119",
@@ -8216,7 +8217,7 @@ dependencies = [
 "reqwest",
 "serde_json",
 "smallvec",
- "spin 0.12.2",
+ "spin 0.12.3",
 "symbolic-demangle",
 "tempfile",
 "thiserror 2.0.20",
@@ -8285,9 +8286,9 @@ dependencies = [

 [[package]]
 name = "quinn-proto"
-version = "0.11.16"
+version = "0.11.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560"
+checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83"
 dependencies = [
 "aws-lc-rs",
 "bytes",
@@ -8553,9 +8554,9 @@ dependencies = [

 [[package]]
 name = "redis"
-version = "1.5.0"
+version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3257df217f7eab0044627a268c9cc6cdb60c0c421c88f83ac41c4e31520b6b84"
+checksum = "e37a4ca5c6ca42aa3e6df2fd32b987a65d32a4c2159a6f3fe0fd1df306a2658f"
 dependencies = [
 "arc-swap",
 "arcstr",
@@ -8567,7 +8568,7 @@ dependencies = [
 "futures-channel",
 "futures-util",
 "itoa",
- "num-bigint 0.4.8",
+ "num-bigint 0.5.1",
 "percent-encoding",
 "pin-project-lite",
 "rustls",
@@ -8868,9 +8869,9 @@ dependencies = [

 [[package]]
 name = "rumqttc-core-next"
-version = "0.33.3"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7d7d9205738dd41a2546e82d27a634d07d8b303dcf7558565ff70caf3ceb0f9c"
+checksum = "249896ab27ed630590971738264baa8f722f18965d2e387c706c40a3c2a572cc"
 dependencies = [
 "async-tungstenite",
 "futures-io",
@@ -8886,18 +8887,18 @@ dependencies = [

 [[package]]
 name = "rumqttc-next"
-version = "0.33.3"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ed1bad2180ff539da671da9a996152a921bc5316eb6d8a9cc3bd441653138b08"
+checksum = "477c9bbfba8f3aecc7aad31c6de2eacb75822efaa18e7aeecb8d3d8e534fbf07"
 dependencies = [
 "rumqttc-v5-next",
 ]

 [[package]]
 name = "rumqttc-v5-next"
-version = "0.33.3"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "229576cbedfa9089f90c17c9454e9429ac1e89cdd223bac5cb39d837593f79bc"
+checksum = "3dfa6ddcc7a7dd5688f9bf78d8f81cb94f367bce56c055d8d94cf81ecb0518bf"
 dependencies = [
 "async-tungstenite",
 "bytes",
@@ -8920,9 +8921,9 @@ dependencies = [

 [[package]]
 name = "russh"
-version = "0.62.6"
+version = "0.62.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b41043523e0edcbd4e31d00903e26f12994f63b21bae9904f7405c1ed92752a5"
+checksum = "9decb68e4e44e1079700e54f17c8f23806ec53d7e0db73ab1c71d9dabc666812"
 dependencies = [
 "aes 0.9.2",
 "aws-lc-rs",
@@ -8945,7 +8946,7 @@ dependencies = [
 "enum_dispatch",
 "flate2",
 "futures",
- "generic-array 1.4.4",
+ "generic-array 1.4.5",
 "getrandom 0.4.3",
 "ghash",
 "hex-literal",
@@ -9491,7 +9492,7 @@ dependencies = [
 "parking_lot",
 "rayon",
 "smallvec",
- "spin 0.12.2",
+ "spin 0.12.3",
 ]

 [[package]]
@@ -10833,7 +10834,7 @@ checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc"
 dependencies = [
 "base16ct 0.2.0",
 "der 0.7.10",
- "generic-array 0.14.9",
+ "generic-array 0.14.7",
 "pkcs8 0.10.2",
 "subtle",
 "zeroize",
@@ -11397,9 +11398,9 @@ checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3"

 [[package]]
 name = "spin"
-version = "0.12.2"
+version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8abadc99fd9c7bbb7d0ca2b31d72a067d0c0dcd7aad25ab8cac71ba91417694b"
+checksum = "0134f9043ed38b087ac4f7d4af44c79e2c9e5094421fe3164f435ce585953b10"
 dependencies = [
 "lock_api",
 ]
@@ -11811,7 +11812,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
 "fastrand",
- "getrandom 0.3.4",
+ "getrandom 0.4.3",
 "once_cell",
 "rustix",
 "windows-sys 0.61.2",
@@ -11975,9 +11976,9 @@ dependencies = [

 [[package]]
 name = "tinystr"
-version = "0.8.3"
+version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8323304221c2a851516f22236c5722a72eaa19749016521d6dff0824447d96d"
+checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643"
 dependencies = [
 "displaydoc",
 "zerovec",
@@ -12651,9 +12652,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"

 [[package]]
 name = "uuid"
-version = "1.24.0"
+version = "1.24.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239"
+checksum = "2cefc03fd367c0c6d4305de1b312cf00248c4114f4a0418ce6a6af769e3b0bd9"
 dependencies = [
 "getrandom 0.4.3",
 "js-sys",
@@ -13168,9 +13169,9 @@ dependencies = [

 [[package]]
 name = "writeable"
-version = "0.6.3"
+version = "0.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
+checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc"

 [[package]]
 name = "x509-cert"
@@ -13350,9 +13351,9 @@ dependencies = [

 [[package]]
 name = "zerotrie"
-version = "0.2.4"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0f9152d31db0792fa83f70fb2f83148effb5c1f5b8c7686c3459e361d9bc20bf"
+checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f"
 dependencies = [
 "displaydoc",
 "yoke",
@@ -13361,9 +13362,9 @@ dependencies = [

 [[package]]
 name = "zerovec"
-version = "0.11.6"
+version = "0.11.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "90f911cbc359ab6af17377d242225f4d75119aec87ea711a880987b18cd7b239"
+checksum = "94b5c6b5976d66c1d703c4fd17d3f5e43c8cedaacf604961b171adc7130896d8"
 dependencies = [
 "yoke",
 "zerofrom",
@@ -13372,13 +13373,13 @@ dependencies = [

 [[package]]
 name = "zerovec-derive"
-version = "0.11.3"
+version = "0.11.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555"
+checksum = "9f212a141d820099d57ffafb9569be9617a6f27d3dc881fbee8fb56642f917a9"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.119",
+ "syn 3.0.3",
 ]

 [[package]]
@@ -228,9 +228,9 @@ atoi = "3.1.0"
 atomic_enum = "0.3.0"
 aws-config = { version = "1.10.1" }
 aws-credential-types = { version = "1.3.0" }
-aws-sdk-kms = { default-features = false, version = "1.114.0" }
-aws-sdk-s3 = { default-features = false, version = "1.141.0" }
-aws-sdk-sts = { default-features = false, version = "1.110.0" }
+aws-sdk-kms = { default-features = false, version = "1.115.0" }
+aws-sdk-s3 = { default-features = false, version = "1.142.0" }
+aws-sdk-sts = { default-features = false, version = "1.111.0" }
 aws-smithy-http-client = { default-features = false, version = "1.3.0" }
 aws-smithy-runtime-api = { version = "1.14.0" }
 aws-smithy-types = { version = "1.6.2" }
@@ -284,8 +284,8 @@ rayon = "1.12.0"
 reed-solomon-erasure = { package = "rustfs-erasure-codec", version = "8.0.2" }
 reed-solomon-simd = "3.1.0"
 regex = { version = "1.13.1" }
-rumqttc = { package = "rumqttc-next", version = "0.33.3" }
-redis = { version = "1.5.0" }
+rumqttc = { package = "rumqttc-next", version = "0.34.0" }
+redis = { version = "1.6.0" }
 rustify = { version = "0.7", default-features = false }
 rustix = { version = "1.1.4" }
 rust-embed = { version = "8.12.0" }
@@ -313,7 +313,7 @@ tracing-subscriber = { version = "0.3.23" }
 transform-stream = "0.3.1"
 url = "2.5.8"
 urlencoding = "2.1.3"
-uuid = { version = "1.24.0" }
+uuid = { version = "1.24.1" }
 vaultrs = { version = "0.8.0" }
 tar = "0.4.46"
 walkdir = "2.5.0"
@@ -341,7 +341,7 @@ libunftp = { version = "0.23.0" }
 unftp-core = "0.1.0"
 suppaftp = { version = "10.0.1" }
 rcgen = { version = "0.14.9", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] }
-russh = { version = "0.62.6" }
+russh = { version = "0.62.7" }
 russh-sftp = "2.4.0"

 # WebDAV
@@ -350,7 +350,7 @@ dav-server = "0.11.0"
 # Performance Analysis and Memory Profiling
 mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" }
 libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] }
-hotpath = { version = "0.23.2", default-features = false }
+hotpath = { version = "0.23.3", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.48" }

@@ -287,6 +287,9 @@ pub enum HealRequestSource {
    Scanner,
    AutoHeal,
    ReadRepair,
+    /// Mission Repair Feed: intents delivered by error paths and replayed
+    /// from the durable MRF journal.
+    Mrf,
 }

 impl HealRequestSource {
@@ -297,6 +300,7 @@ impl HealRequestSource {
            Self::Scanner => "scanner",
            Self::AutoHeal => "auto_heal",
            Self::ReadRepair => "read_repair",
+            Self::Mrf => "mrf",
        }
    }
 }
@@ -17,6 +17,7 @@ pub mod globals;
 pub mod heal_channel;
 pub mod last_minute;
 pub mod metrics;
+pub mod mrf_channel;
 mod readiness;
 pub mod table_catalog;
 pub mod trace_bus;
@@ -0,0 +1,203 @@
+// 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.
+
+//! Mission Repair Feed (MRF) intent channel.
+//!
+//! Producers on error paths (read decode failure, scanner metadata
+//! corruption, partial-write recovery) hand a lightweight [`MrfIntent`] to the
+//! heal crate through a global bounded channel. Delivery is strictly
+//! non-blocking: `try_send_mrf_intent` never awaits and drops the intent
+//! (counting it) when the channel is full or uninitialized — losing one heal
+//! hint is always preferred over stalling an IO path. Durable replay of
+//! unconsumed intents is the consumer's job (see `rustfs-heal`
+//! `heal::mrf_queue`), mirroring MinIO's `.heal/mrf/list.bin`.
+
+use std::sync::{
+    Arc, OnceLock,
+    atomic::{AtomicBool, Ordering},
+};
+use tokio::sync::mpsc;
+use uuid::Uuid;
+
+/// Bounded capacity of the global MRF channel. Backpressure is resolved by
+/// dropping (and counting) intents, never by blocking the producer.
+const MRF_CHANNEL_CAPACITY: usize = 8192;
+
+/// Why an intent was produced. Drives the heal priority mapping on the
+/// consumer side (DecodeFailure -> Urgent, MetadataCorruption -> High,
+/// PartialWrite -> Normal).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum MrfKind {
+    /// Erasure decode failed while serving a read (read path).
+    DecodeFailure,
+    /// Scanner classified object metadata as corrupt.
+    MetadataCorruption,
+    /// A write left the object with fewer committed shards than the set size.
+    PartialWrite,
+}
+
+impl MrfKind {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            MrfKind::DecodeFailure => "decode-failure",
+            MrfKind::MetadataCorruption => "metadata-corruption",
+            MrfKind::PartialWrite => "partial-write",
+        }
+    }
+}
+
+/// One repair intent. Kept deliberately small so the in-memory queue and the
+/// journal stay bounded; `bucket`/`object` are `Arc<str>` so re-arming an
+/// intent never re-allocates the strings.
+#[derive(Clone, Debug)]
+pub struct MrfIntent {
+    pub bucket: Arc<str>,
+    pub object: Arc<str>,
+    /// Version the intent targets, as raw UUID bytes.
+    pub version_id: Option<[u8; 16]>,
+    pub kind: MrfKind,
+    pub enqueued_at_ms: u64,
+    /// Times this intent has already been offered to the heal manager.
+    /// Dropped by the consumer once it reaches `MRF_MAX_ATTEMPTS`.
+    pub attempts: u8,
+}
+
+/// Consumer-side retry ceiling before an intent is given up on.
+pub const MRF_MAX_ATTEMPTS: u8 = 3;
+
+impl MrfIntent {
+    /// Rough in-memory footprint used by the queue's byte budget.
+    pub fn estimated_bytes(&self) -> usize {
+        // Struct + strings + version bytes; buckets and objects are usually
+        // far below this bound, so rounding up keeps the budget conservative.
+        64 + self.bucket.len() + self.object.len()
+    }
+}
+
+static GLOBAL_MRF_SENDER: OnceLock<mpsc::Sender<MrfIntent>> = OnceLock::new();
+
+/// Delivery kill-switch, set from `RUSTFS_HEAL_MRF_ENABLE`. Producers check
+/// this before touching the channel so the disabled path stays allocation- and
+/// sync-free.
+static MRF_DELIVERY_ENABLED: AtomicBool = AtomicBool::new(true);
+
+/// Override delivery (used at heal-runtime startup from configuration).
+pub fn set_mrf_delivery_enabled(enabled: bool) {
+    MRF_DELIVERY_ENABLED.store(enabled, Ordering::Relaxed);
+}
+
+/// Whether producers currently deliver intents.
+pub fn mrf_delivery_enabled() -> bool {
+    MRF_DELIVERY_ENABLED.load(Ordering::Relaxed)
+}
+
+/// Create the global MRF channel and return the consumer half. Fails if the
+/// channel is already initialized (the heal runtime is a singleton).
+pub fn init_mrf_channel() -> Result<mpsc::Receiver<MrfIntent>, &'static str> {
+    let (sender, receiver) = mpsc::channel(MRF_CHANNEL_CAPACITY);
+    GLOBAL_MRF_SENDER
+        .set(sender)
+        .map_err(|_| "MRF channel sender already initialized")?;
+    Ok(receiver)
+}
+
+/// Best-effort, non-blocking intent delivery from an error path.
+///
+/// Returns `true` when the intent was accepted into the channel. `false`
+/// means the intent was dropped (feature disabled, channel not yet
+/// initialized, or channel full) — callers must not retry or await; the
+/// existing read-repair / scanner heal paths remain the safety net.
+///
+/// This runs on IO error paths, so it stays synchronous and cheap: one
+/// bounded allocation for the two `Arc<str>` handles plus the channel slot.
+pub fn try_send_mrf_intent(kind: MrfKind, bucket: &str, object: &str, version_id: Option<Uuid>) -> bool {
+    if !mrf_delivery_enabled() {
+        return false;
+    }
+    let Some(sender) = GLOBAL_MRF_SENDER.get() else {
+        return false;
+    };
+    let intent = MrfIntent {
+        bucket: Arc::from(bucket),
+        object: Arc::from(object),
+        version_id: version_id.map(|vid| *vid.as_bytes()),
+        kind,
+        enqueued_at_ms: unix_now_ms(),
+        attempts: 0,
+    };
+    sender.try_send(intent).is_ok()
+}
+
+fn unix_now_ms() -> u64 {
+    // Kept trivial: the timestamp is diagnostic metadata only; wall-clock
+    // failure would be a bug rather than something to handle here.
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn intents_estimate_is_conservative() {
+        let intent = MrfIntent {
+            bucket: Arc::from("bucket"),
+            object: Arc::from("object"),
+            version_id: Some([0u8; 16]),
+            kind: MrfKind::DecodeFailure,
+            enqueued_at_ms: 0,
+            attempts: 0,
+        };
+        assert!(intent.estimated_bytes() >= intent.bucket.len() + intent.object.len());
+    }
+
+    #[tokio::test]
+    async fn try_send_delivers_and_respects_capacity() {
+        let mut receiver = init_mrf_channel().expect("first initialization should succeed");
+        assert!(init_mrf_channel().is_err(), "double initialization must fail");
+
+        assert!(try_send_mrf_intent(MrfKind::DecodeFailure, "b", "o", Some(Uuid::nil())));
+        let intent = receiver.recv().await.expect("intent should arrive");
+        assert_eq!(intent.kind, MrfKind::DecodeFailure);
+        assert_eq!(intent.bucket.as_ref(), "b");
+
+        // Disable delivery: producers become no-ops.
+        set_mrf_delivery_enabled(false);
+        assert!(!try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None));
+        set_mrf_delivery_enabled(true);
+
+        // Fill the bounded channel past capacity: excess intents are dropped,
+        // never blocking.
+        let mut accepted = 0;
+        for _ in 0..(MRF_CHANNEL_CAPACITY + 64) {
+            if try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None) {
+                accepted += 1;
+            }
+        }
+        assert_eq!(accepted, MRF_CHANNEL_CAPACITY);
+    }
+
+    #[test]
+    fn try_send_without_channel_is_false() {
+        // This test may run after the tokio test above in the same process;
+        // the singleton semantics make a clean "uninitialized" case hard, so
+        // assert the flag-off behavior only.
+        set_mrf_delivery_enabled(false);
+        assert!(!try_send_mrf_intent(MrfKind::MetadataCorruption, "b", "o", None));
+        set_mrf_delivery_enabled(true);
+    }
+}
@@ -177,3 +177,31 @@ pub const DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT: usize = 80;

 /// Default foreground pressure recheck delay for heal scheduler, in milliseconds.
 pub const DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS: u64 = 250;
+
+/// Environment variable that toggles the MRF (mission repair feed) intent
+/// pipeline: error paths deliver repair intents to the heal runtime, and
+/// unconsumed intents are replayed from the durable journal after a restart.
+pub const ENV_HEAL_MRF_ENABLE: &str = "RUSTFS_HEAL_MRF_ENABLE";
+
+/// Environment variable for the MRF in-memory queue capacity (intent count).
+pub const ENV_HEAL_MRF_QUEUE_SIZE: &str = "RUSTFS_HEAL_MRF_QUEUE_SIZE";
+
+/// Environment variable for the MRF journal byte budget. The journal is
+/// compacted once its on-disk size crosses this bound.
+pub const ENV_HEAL_MRF_JOURNAL_MAX_BYTES: &str = "RUSTFS_HEAL_MRF_JOURNAL_MAX_BYTES";
+
+/// Environment variable for the MRF journal replay batch size (intents per
+/// replay push round).
+pub const ENV_HEAL_MRF_REPLAY_BATCH: &str = "RUSTFS_HEAL_MRF_REPLAY_BATCH";
+
+/// Default behavior keeps the MRF intent pipeline enabled.
+pub const DEFAULT_HEAL_MRF_ENABLE: bool = true;
+
+/// Default MRF queue capacity (matches MinIO's 100k MRF list ceiling).
+pub const DEFAULT_HEAL_MRF_QUEUE_SIZE: usize = 100_000;
+
+/// Default MRF journal byte budget (8 MiB), mirroring the channel payload cap.
+pub const DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES: usize = 8 * 1024 * 1024;
+
+/// Default MRF replay batch size.
+pub const DEFAULT_HEAL_MRF_REPLAY_BATCH: usize = 256;
@@ -234,6 +234,31 @@ pub const ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP: &str = "RUSTFS_OBJECT_DISK_WRITE_A
 /// Default absolute per-object erasure write cap in seconds (`0` = disabled).
 pub const DEFAULT_OBJECT_DISK_WRITE_ABSOLUTE_CAP: u64 = 0;

+/// Enable foreground PutObject request admission.
+///
+/// This is an experimental, default-off foreground write backpressure gate for
+/// strict commit tail investigations. When disabled, PUTs follow the legacy
+/// path and only the existing request counters are updated.
+pub const ENV_PUT_FOREGROUND_ADMISSION_ENABLE: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_ENABLE";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE: bool = false;
+
+/// Maximum foreground PutObject requests admitted concurrently per process.
+///
+/// The limit is used only when [`ENV_PUT_FOREGROUND_ADMISSION_ENABLE`] is true.
+/// A value of `0` disables the gate even when the enable flag is present, so a
+/// partially configured rollout cannot reject every PUT.
+pub const ENV_PUT_FOREGROUND_ADMISSION_LIMIT: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_LIMIT";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT: usize = 0;
+
+/// Time in milliseconds a foreground PutObject waits for an admission permit.
+///
+/// Once this timeout expires the request fails before body ingest/storage
+/// mutation with S3 `SlowDown`/503. `0` means fail fast when the limit is full.
+pub const ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 0;
+
+const _: () = assert!(!DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE);
+
 /// Environment variable for minimum GetObject timeout in seconds.
 ///
 /// When dynamic timeout calculation is enabled, this is the minimum timeout
@@ -5845,6 +5845,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {

    #[tracing::instrument(skip(self))]
    async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> {
+        // MRF journal intent: partial-write recovery must survive a restart
+        // (HS-01); the heal request below remains the in-memory fast path.
+        rustfs_common::mrf_channel::try_send_mrf_intent(
+            rustfs_common::mrf_channel::MrfKind::PartialWrite,
+            bucket,
+            object,
+            uuid::Uuid::try_parse(version_id).ok(),
+        );
        let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
            bucket.to_string(),
            Some(object.to_string()),
@@ -1077,6 +1077,15 @@ impl SetDisks {
                            "Recoverable decode error triggered read repair"
                        );
                        let version_id = fi.version_id.as_ref().map(ToString::to_string);
+                        // MRF journal intent: keeps a durable Urgent ECDecode
+                        // request alive across restarts even when the in-memory
+                        // read-repair request is dropped or lost (HS-01).
+                        rustfs_common::mrf_channel::try_send_mrf_intent(
+                            rustfs_common::mrf_channel::MrfKind::DecodeFailure,
+                            bucket,
+                            object,
+                            fi.version_id,
+                        );
                        submit_read_repair_heal(
                            bucket,
                            object,
@@ -89,6 +89,8 @@ async-trait = { workspace = true }
 futures = { workspace = true }
 metrics = { workspace = true }
 base64 = { workspace = true }
+bytes = { workspace = true }
+crc-fast = { workspace = true }

 [dev-dependencies]
 serde_json = { workspace = true, features = ["raw_value"] }
@@ -612,7 +612,8 @@ impl HealChannelProcessor {
            HealRequestSource::Admin
            | HealRequestSource::AutoHeal
            | HealRequestSource::Internal
-            | HealRequestSource::ReadRepair => true,
+            | HealRequestSource::ReadRepair
+            | HealRequestSource::Mrf => true,
        });

        // Build HealOptions with all available fields
@@ -270,6 +270,8 @@ pub struct HealSourceCounts {
    pub auto_heal: u64,
    pub internal: u64,
    pub read_repair: u64,
+    #[serde(default)]
+    pub mrf: u64,
 }

 impl HealSourceCounts {
@@ -280,6 +282,7 @@ impl HealSourceCounts {
            HealRequestSource::AutoHeal => self.auto_heal += 1,
            HealRequestSource::Internal => self.internal += 1,
            HealRequestSource::ReadRepair => self.read_repair += 1,
+            HealRequestSource::Mrf => self.mrf += 1,
        }
    }
 }
@@ -16,6 +16,7 @@ pub mod channel;
 pub mod erasure_healer;
 pub mod event;
 pub mod manager;
+pub mod mrf_queue;
 pub mod progress;
 pub(crate) mod replacement_readiness;
 pub mod resume;
@@ -0,0 +1,682 @@
+// 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.
+
+//! Mission Repair Feed (MRF) queue, journal, and consumer.
+//!
+//! Intents arriving on the global channel (see `rustfs_common::mrf_channel`)
+//! are buffered in a bounded in-memory queue, translated into prioritized
+//! heal requests, and — while they are not yet accepted by the heal manager —
+//! mirrored into a durable journal so a crash or restart can replay them.
+//! This is the RustFS counterpart of MinIO's `.heal/mrf/list.bin` replay,
+//! layered on top of (not replacing) read-repair and scanner heal.
+//!
+//! Durability model: the journal is a snapshot of the *unaccepted* pending
+//! set, rewritten on a group-commit cadence (every flush interval or flush
+//! threshold new intents). A rewrite is atomic at the record level only — a
+//! torn tail simply truncates during replay because every record carries its
+//! own CRC32. Losing the last flush window (≤500 ms) is acceptable: replayed
+//! duplicates are merged by the manager's dedup key, and read-repair remains
+//! the safety net.
+
+use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
+use crate::heal::manager::HealManager;
+use metrics::{counter, gauge};
+use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
+use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIntent};
+use std::collections::VecDeque;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::mpsc;
+use uuid::Uuid;
+
+use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealType};
+
+/// Journal location inside the metadata bucket, following the resume-state
+/// layout.
+pub(crate) const MRF_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal.bin";
+
+/// Record format tag.
+const MRF_JOURNAL_FORMAT: u8 = 1;
+/// Record layout version.
+const MRF_JOURNAL_VERSION: u8 = 1;
+
+/// Fixed header size: format, version, kind, attempts, enqueued_at_ms,
+/// has_version flag.
+const MRF_RECORD_FIXED_HEAD: usize = 1 + 1 + 1 + 1 + 8 + 1;
+
+#[derive(Debug, Clone)]
+pub(crate) struct MrfConsumerConfig {
+    /// In-memory queue capacity in intents.
+    pub queue_capacity: usize,
+    /// Journal byte budget; a pending snapshot above this bound is rejected
+    /// oldest-first so the journal can never grow unbounded.
+    pub journal_max_bytes: usize,
+    /// How many journal intents to re-arm per replay round.
+    pub replay_batch: usize,
+    /// Group-commit cadence for the journal snapshot.
+    pub flush_interval: Duration,
+    /// New intents between flushes that force an early snapshot.
+    pub flush_threshold: usize,
+    /// Backoff after the heal manager reports a full admission.
+    pub admission_backoff: Duration,
+}
+
+impl Default for MrfConsumerConfig {
+    fn default() -> Self {
+        Self {
+            queue_capacity: rustfs_utils::get_env_usize(
+                rustfs_config::ENV_HEAL_MRF_QUEUE_SIZE,
+                rustfs_config::DEFAULT_HEAL_MRF_QUEUE_SIZE,
+            ),
+            journal_max_bytes: rustfs_utils::get_env_usize(
+                rustfs_config::ENV_HEAL_MRF_JOURNAL_MAX_BYTES,
+                rustfs_config::DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES,
+            ),
+            replay_batch: rustfs_utils::get_env_usize(
+                rustfs_config::ENV_HEAL_MRF_REPLAY_BATCH,
+                rustfs_config::DEFAULT_HEAL_MRF_REPLAY_BATCH,
+            ),
+            flush_interval: Duration::from_millis(500),
+            flush_threshold: 1000,
+            admission_backoff: Duration::from_secs(5),
+        }
+    }
+}
+
+/// Bounded pending set with count and byte ceilings. Overflow drops the
+/// incoming intent (never a resident one) and counts the loss.
+pub(crate) struct MrfQueue {
+    pending: VecDeque<MrfIntent>,
+    bytes: usize,
+    capacity: usize,
+    byte_budget: usize,
+}
+
+impl MrfQueue {
+    pub(crate) fn new(capacity: usize, byte_budget: usize) -> Self {
+        Self {
+            pending: VecDeque::new(),
+            bytes: 0,
+            capacity,
+            byte_budget,
+        }
+    }
+
+    /// Returns `false` (after counting) when either ceiling would be crossed.
+    pub(crate) fn try_push(&mut self, intent: MrfIntent) -> bool {
+        let cost = intent.estimated_bytes();
+        if self.pending.len() >= self.capacity || self.bytes + cost > self.byte_budget {
+            counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1);
+            return false;
+        }
+        self.bytes += cost;
+        self.pending.push_back(intent);
+        true
+    }
+
+    pub(crate) fn pop_front(&mut self) -> Option<MrfIntent> {
+        let intent = self.pending.pop_front()?;
+        self.bytes = self.bytes.saturating_sub(intent.estimated_bytes());
+        Some(intent)
+    }
+
+    pub(crate) fn push_back(&mut self, intent: MrfIntent) {
+        self.bytes += intent.estimated_bytes();
+        self.pending.push_back(intent);
+    }
+
+    pub(crate) fn depth(&self) -> usize {
+        self.pending.len()
+    }
+
+    pub(crate) fn bytes(&self) -> usize {
+        self.bytes
+    }
+
+    pub(crate) fn intents(&self) -> impl Iterator<Item = &MrfIntent> {
+        self.pending.iter()
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Journal record codec
+// ---------------------------------------------------------------------------
+
+/// Append one encoded record to `out`.
+pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec<u8>) {
+    let start = out.len();
+    out.push(MRF_JOURNAL_FORMAT);
+    out.push(MRF_JOURNAL_VERSION);
+    out.push(match intent.kind {
+        rustfs_common::mrf_channel::MrfKind::DecodeFailure => 1,
+        rustfs_common::mrf_channel::MrfKind::MetadataCorruption => 2,
+        rustfs_common::mrf_channel::MrfKind::PartialWrite => 3,
+    });
+    out.push(intent.attempts);
+    out.extend_from_slice(&intent.enqueued_at_ms.to_le_bytes());
+    match intent.version_id {
+        Some(bytes) => {
+            out.push(1);
+            out.extend_from_slice(&bytes);
+        }
+        None => out.push(0),
+    }
+    out.extend_from_slice(&(intent.bucket.len() as u32).to_le_bytes());
+    out.extend_from_slice(&(intent.object.len() as u32).to_le_bytes());
+    out.extend_from_slice(intent.bucket.as_bytes());
+    out.extend_from_slice(intent.object.as_bytes());
+    let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+    hasher.update(&out[start..]);
+    out.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes());
+}
+
+fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
+    if data.len() < MRF_RECORD_FIXED_HEAD + 8 {
+        return None;
+    }
+    if data[0] != MRF_JOURNAL_FORMAT || data[1] != MRF_JOURNAL_VERSION {
+        return None;
+    }
+    let kind = match data[2] {
+        1 => rustfs_common::mrf_channel::MrfKind::DecodeFailure,
+        2 => rustfs_common::mrf_channel::MrfKind::MetadataCorruption,
+        3 => rustfs_common::mrf_channel::MrfKind::PartialWrite,
+        _ => return None,
+    };
+    let attempts = data[3];
+    let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().expect("slice length checked"));
+    let has_version = data[12] != 0;
+    let mut cursor = MRF_RECORD_FIXED_HEAD;
+    let version_id = if has_version {
+        if data.len() < cursor + 16 {
+            return None;
+        }
+        let bytes: [u8; 16] = data[cursor..cursor + 16].try_into().expect("slice length checked");
+        cursor += 16;
+        Some(bytes)
+    } else {
+        None
+    };
+    if data.len() < cursor + 8 {
+        return None;
+    }
+    let bucket_len = u32::from_le_bytes(data[cursor..cursor + 4].try_into().expect("slice length checked")) as usize;
+    let object_len = u32::from_le_bytes(data[cursor + 4..cursor + 8].try_into().expect("slice length checked")) as usize;
+    cursor += 8;
+    let body_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?;
+    let record_end = body_end.checked_add(4)?;
+    if data.len() < record_end {
+        return None;
+    }
+    let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+    hasher.update(&data[..body_end]);
+    if (hasher.finalize() as u32) != u32::from_le_bytes(data[body_end..record_end].try_into().expect("slice length checked")) {
+        return None;
+    }
+    let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?);
+    let object = std::sync::Arc::from(std::str::from_utf8(&data[cursor + bucket_len..body_end]).ok()?);
+    Some((
+        MrfIntent {
+            bucket,
+            object,
+            version_id,
+            kind,
+            enqueued_at_ms,
+            attempts,
+        },
+        record_end,
+    ))
+}
+
+/// Decode a whole journal, stopping at the first torn or corrupt record.
+/// Returns the decoded intents and the number of trailing bytes discarded.
+pub(crate) fn decode_journal(data: &[u8]) -> (Vec<MrfIntent>, usize) {
+    let mut intents = Vec::new();
+    let mut cursor = 0usize;
+    while cursor < data.len() {
+        match decode_one(&data[cursor..]) {
+            Some((intent, consumed)) => {
+                intents.push(intent);
+                cursor += consumed;
+            }
+            None => break,
+        }
+    }
+    let truncated = data.len() - cursor;
+    (intents, truncated)
+}
+
+// ---------------------------------------------------------------------------
+// Journal disk IO (all local disks, first successful read wins)
+// ---------------------------------------------------------------------------
+
+async fn journal_disks() -> Vec<DiskStore> {
+    let map = local_disk_map_read().await;
+    map.values().flatten().cloned().collect()
+}
+
+async fn read_journal() -> Option<Vec<u8>> {
+    for disk in journal_disks().await {
+        match disk.read_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH).await {
+            Ok(bytes) => return Some(bytes.to_vec()),
+            Err(_) => continue,
+        }
+    }
+    None
+}
+
+async fn write_journal(data: &[u8]) {
+    let payload = bytes::Bytes::copy_from_slice(data);
+    for disk in journal_disks().await {
+        if let Err(err) = disk
+            .write_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH, payload.clone())
+            .await
+        {
+            warn_mrf_journal_write(&err);
+        }
+    }
+    if !data.is_empty() {
+        counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
+    }
+    gauge!("rustfs_heal_mrf_journal_bytes").set(data.len() as f64);
+}
+
+async fn delete_journal() {
+    for disk in journal_disks().await {
+        let _ = disk
+            .delete(
+                super::RUSTFS_META_BUCKET,
+                MRF_JOURNAL_PATH,
+                crate::heal::storage_api::owner::EcstoreDeleteOptions::default(),
+            )
+            .await;
+    }
+}
+
+fn warn_mrf_journal_write(err: &super::DiskError) {
+    tracing::warn!(
+        target: "rustfs::heal::mrf",
+        error = %err,
+        "MRF journal write failed; unconsumed intents may be lost on restart"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// Consumer
+// ---------------------------------------------------------------------------
+
+/// Translate an intent into the prioritized heal request the issue specifies:
+/// decode failures go Urgent ECDecode, metadata corruption goes High
+/// Metadata, partial writes go Normal object heal.
+pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest {
+    let bucket = intent.bucket.to_string();
+    let object = intent.object.to_string();
+    let version_id = intent.version_id.map(|bytes| Uuid::from_bytes(bytes).to_string());
+    let (heal_type, priority) = match intent.kind {
+        rustfs_common::mrf_channel::MrfKind::DecodeFailure => (
+            HealType::ECDecode {
+                bucket,
+                object,
+                version_id,
+            },
+            HealPriority::Urgent,
+        ),
+        rustfs_common::mrf_channel::MrfKind::MetadataCorruption => (HealType::Metadata { bucket, object }, HealPriority::High),
+        rustfs_common::mrf_channel::MrfKind::PartialWrite => (
+            HealType::Object {
+                bucket,
+                object,
+                version_id,
+            },
+            HealPriority::Normal,
+        ),
+    };
+    let mut request = HealRequest::new(heal_type, HealOptions::default(), priority);
+    request.source = rustfs_common::heal_channel::HealRequestSource::Mrf;
+    request
+}
+
+struct MrfRuntime {
+    queue: MrfQueue,
+    config: MrfConsumerConfig,
+    new_since_flush: usize,
+    /// True while a journal snapshot exists on disk that no longer reflects
+    /// an all-consumed pending set; the next idle tick removes it (MinIO
+    /// deletes its `list.bin` after replay for the same reason).
+    journal_on_disk: bool,
+    /// Earliest instant a full-admission retry may proceed.
+    backoff_until: Option<tokio::time::Instant>,
+}
+
+impl MrfRuntime {
+    fn record_accept(&mut self) {
+        // Accepted intents leave the pending set; the next flush persists the
+        // smaller snapshot, which is the journal's compaction.
+    }
+
+    fn snapshot(&self) -> Vec<u8> {
+        let mut buf = Vec::new();
+        for intent in self.queue.intents() {
+            encode_intent(intent, &mut buf);
+        }
+        buf
+    }
+
+    async fn flush(&mut self) {
+        write_journal(&self.snapshot()).await;
+        self.new_since_flush = 0;
+        self.journal_on_disk = true;
+    }
+
+    /// Drain pending intents into the heal manager until it is full, the
+    /// queue empties, or attempts are exhausted.
+    async fn dispatch(&mut self, manager: &HealManager) {
+        if let Some(until) = self.backoff_until {
+            if tokio::time::Instant::now() < until {
+                return;
+            }
+            self.backoff_until = None;
+        }
+        while let Some(mut intent) = self.queue.pop_front() {
+            let request = build_heal_request(&intent);
+            match manager.submit_heal_request(request).await {
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => self.record_accept(),
+                Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
+                    intent.attempts = intent.attempts.saturating_add(1);
+                    if intent.attempts >= MRF_MAX_ATTEMPTS {
+                        counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1);
+                        continue;
+                    }
+                    self.queue.push_back(intent);
+                    self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff);
+                    break;
+                }
+                Ok(HealAdmissionResult::Dropped(_)) => {
+                    counter!("rustfs_heal_mrf_dropped_total", "reason" => "admission_policy").increment(1);
+                }
+                Err(_) => {
+                    intent.attempts = intent.attempts.saturating_add(1);
+                    if intent.attempts >= MRF_MAX_ATTEMPTS {
+                        counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1);
+                        continue;
+                    }
+                    self.queue.push_back(intent);
+                    self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff);
+                    break;
+                }
+            }
+        }
+        gauge!("rustfs_heal_mrf_queue_depth").set(self.queue.depth() as f64);
+        gauge!("rustfs_heal_mrf_queue_bytes").set(self.queue.bytes() as f64);
+    }
+}
+
+/// Initialize the global MRF channel (honoring `RUSTFS_HEAL_MRF_ENABLE`) and
+/// spawn the consumer task. Called once from the heal runtime bootstrap right
+/// after the manager started; a disabled feature or a double call is a no-op.
+/// Public for integration tests that drive the real consumer loop.
+pub fn spawn_mrf_consumer(manager: Arc<HealManager>) {
+    let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_HEAL_MRF_ENABLE, rustfs_config::DEFAULT_HEAL_MRF_ENABLE);
+    rustfs_common::mrf_channel::set_mrf_delivery_enabled(enabled);
+    if !enabled {
+        tracing::info!(
+            target: "rustfs::heal::mrf",
+            "MRF intent pipeline disabled by configuration; producers will not deliver"
+        );
+        return;
+    }
+    let receiver = match rustfs_common::mrf_channel::init_mrf_channel() {
+        Ok(receiver) => receiver,
+        Err(err) => {
+            tracing::warn!(
+                target: "rustfs::heal::mrf",
+                error = err,
+                "MRF channel initialization failed; intents will be dropped at producers"
+            );
+            return;
+        }
+    };
+    tokio::spawn(async move {
+        run_mrf_consumer(manager, receiver).await;
+    });
+    tracing::info!(target: "rustfs::heal::mrf", "MRF intent consumer started");
+}
+
+/// Replay the durable journal into a fresh pending queue and submit whatever
+/// it armed. Returns the number of intact intents replayed. Duplicates are
+/// merged by the manager's dedup key; the journal file is removed once read
+/// (torn tails truncate via the per-record CRC). Public for integration tests;
+/// the live consumer invokes this through [`replay_into`] at startup.
+pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
+    let config = MrfConsumerConfig::default();
+    let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
+    let mut backoff_until: Option<tokio::time::Instant> = None;
+    replay_into(manager, &mut queue, &mut backoff_until).await
+}
+
+/// Shared replay core: read + decode + re-arm + delete, then drain what fits.
+async fn replay_into(
+    manager: &Arc<HealManager>,
+    queue: &mut MrfQueue,
+    backoff_until: &mut Option<tokio::time::Instant>,
+) -> usize {
+    let Some(data) = read_journal().await else {
+        return 0;
+    };
+    let (intents, truncated) = decode_journal(&data);
+    if truncated > 0 {
+        tracing::warn!(
+            target: "rustfs::heal::mrf",
+            truncated_bytes = truncated,
+            "MRF journal had a torn tail; truncated records were discarded"
+        );
+    }
+    counter!("rustfs_heal_mrf_replayed_total").increment(intents.len() as u64);
+    let replayed = intents.len();
+    for intent in intents {
+        queue.try_push(intent);
+    }
+    delete_journal().await;
+
+    // Drain the replayed intents immediately; whatever the manager refuses
+    // stays armed in `queue` for the consumer's retry loop.
+    if backoff_until.is_none() {
+        while let Some(mut intent) = queue.pop_front() {
+            let request = build_heal_request(&intent);
+            match manager.submit_heal_request(request).await {
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
+                Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
+                    intent.attempts = intent.attempts.saturating_add(1);
+                    if intent.attempts < MRF_MAX_ATTEMPTS {
+                        queue.push_back(intent);
+                        *backoff_until = Some(tokio::time::Instant::now());
+                    }
+                    break;
+                }
+                Ok(HealAdmissionResult::Dropped(_)) | Err(_) => {}
+            }
+        }
+    }
+    replayed
+}
+
+/// Replay the journal, then keep draining the channel into the heal manager
+/// while persisting the pending snapshot.
+async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receiver<MrfIntent>) {
+    let config = MrfConsumerConfig::default();
+    let mut runtime = MrfRuntime {
+        queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
+        config: config.clone(),
+        new_since_flush: 0,
+        journal_on_disk: false,
+        backoff_until: None,
+    };
+
+    // Replay: read the journal, re-arm intents (duplicates are merged by the
+    // manager's dedup key), then drop the file so the next flush starts clean.
+    replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await;
+
+    let mut flush_tick = tokio::time::interval(runtime.config.flush_interval);
+    flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    let mut batch: Vec<MrfIntent> = Vec::with_capacity(runtime.config.replay_batch);
+
+    loop {
+        tokio::select! {
+            received = receiver.recv_many(&mut batch, runtime.config.replay_batch) => {
+                if received == 0 {
+                    // Channel closed: flush once more and stop.
+                    runtime.flush().await;
+                    tracing::info!(
+                        target: "rustfs::heal::mrf",
+                        "MRF channel closed; consumer stopped after final flush"
+                    );
+                    return;
+                }
+                for intent in batch.drain(..) {
+                    runtime.queue.try_push(intent);
+                    runtime.new_since_flush += 1;
+                }
+                runtime.dispatch(manager.as_ref()).await;
+                if runtime.new_since_flush >= runtime.config.flush_threshold {
+                    runtime.flush().await;
+                }
+            }
+            _ = flush_tick.tick() => {
+                if runtime.new_since_flush > 0 || runtime.queue.depth() > 0 {
+                    runtime.flush().await;
+                    runtime.dispatch(manager.as_ref()).await;
+                } else if runtime.journal_on_disk {
+                    // All intents consumed: remove the journal so a restart
+                    // replays nothing (mirrors MinIO's post-replay unlink).
+                    delete_journal().await;
+                    runtime.journal_on_disk = false;
+                    gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
+                }
+                gauge!("rustfs_heal_mrf_queue_depth").set(runtime.queue.depth() as f64);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rustfs_common::mrf_channel::{MrfIntent, MrfKind};
+    use std::sync::Arc as StdArc;
+
+    fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent {
+        MrfIntent {
+            bucket: StdArc::from(bucket),
+            object: StdArc::from(object),
+            version_id: Some([7u8; 16]),
+            kind: MrfKind::DecodeFailure,
+            enqueued_at_ms: 1_700_000_000_000,
+            attempts,
+        }
+    }
+
+    #[test]
+    fn queue_enforces_count_and_byte_ceilings() {
+        let mut queue = MrfQueue::new(2, usize::MAX);
+        assert!(queue.try_push(intent("b", "o", 0)));
+        assert!(queue.try_push(intent("b", "o", 0)));
+        assert!(!queue.try_push(intent("b", "o", 0)), "count ceiling must drop");
+
+        let mut tiny = MrfQueue::new(usize::MAX, intent("bucket", "object", 0).estimated_bytes());
+        assert!(tiny.try_push(intent("bucket", "object", 0)));
+        assert!(
+            !tiny.try_push(intent("bucket", "object", 0)),
+            "byte budget must drop before the second intent fits"
+        );
+    }
+
+    #[test]
+    fn journal_roundtrip_preserves_intents() {
+        let intents = vec![
+            intent("bucket-a", "object/a", 0),
+            intent("bucket-b", "object/b", 2),
+            MrfIntent {
+                bucket: StdArc::from("bucket-c"),
+                object: StdArc::from("object/c"),
+                version_id: None,
+                kind: MrfKind::MetadataCorruption,
+                enqueued_at_ms: 5,
+                attempts: 1,
+            },
+        ];
+        let mut buf = Vec::new();
+        for intent in &intents {
+            encode_intent(intent, &mut buf);
+        }
+        let (decoded, truncated) = decode_journal(&buf);
+        assert_eq!(truncated, 0);
+        assert_eq!(decoded.len(), intents.len());
+        for (left, right) in decoded.iter().zip(intents.iter()) {
+            assert_eq!(left.bucket, right.bucket);
+            assert_eq!(left.object, right.object);
+            assert_eq!(left.version_id, right.version_id);
+            assert_eq!(left.kind, right.kind);
+            assert_eq!(left.attempts, right.attempts);
+        }
+    }
+
+    #[test]
+    fn journal_torn_tail_is_truncated() {
+        let mut buf = Vec::new();
+        encode_intent(&intent("b", "o", 0), &mut buf);
+        let mut torn = buf.clone();
+        torn.extend_from_slice(&buf[..buf.len() / 2]);
+
+        let (decoded, truncated) = decode_journal(&torn);
+        assert_eq!(decoded.len(), 1, "the intact record must survive");
+        assert!(truncated > 0, "the partial tail must be discarded");
+
+        // A corrupted body (CRC mismatch) also truncates from that record on.
+        let mut corrupt = buf.clone();
+        let mid = MRF_RECORD_FIXED_HEAD + 4;
+        corrupt[mid] ^= 0xff;
+        let (decoded, truncated) = decode_journal(&corrupt);
+        assert!(decoded.is_empty());
+        assert_eq!(truncated, corrupt.len());
+    }
+
+    #[test]
+    fn heal_request_mapping_follows_priority_matrix() {
+        let decode = build_heal_request(&intent("b", "o", 0));
+        assert!(matches!(decode.heal_type, HealType::ECDecode { .. }));
+        assert_eq!(decode.priority, HealPriority::Urgent);
+
+        let metadata = build_heal_request(&MrfIntent {
+            bucket: StdArc::from("b"),
+            object: StdArc::from("o"),
+            version_id: None,
+            kind: MrfKind::MetadataCorruption,
+            enqueued_at_ms: 0,
+            attempts: 0,
+        });
+        assert!(matches!(metadata.heal_type, HealType::Metadata { .. }));
+        assert_eq!(metadata.priority, HealPriority::High);
+
+        let partial = build_heal_request(&MrfIntent {
+            bucket: StdArc::from("b"),
+            object: StdArc::from("o"),
+            version_id: None,
+            kind: MrfKind::PartialWrite,
+            enqueued_at_ms: 0,
+            attempts: 0,
+        });
+        assert!(matches!(partial.heal_type, HealType::Object { .. }));
+        assert_eq!(partial.priority, HealPriority::Normal);
+    }
+}
@@ -158,6 +158,10 @@ pub async fn init_heal_manager_with_workload_provider(
            return Err(err);
        }

+        // Start the MRF intent consumer (error-path repair intents + durable
+        // journal replay) now that the manager can accept submissions.
+        heal::mrf_queue::spawn_mrf_consumer(heal_manager.clone());
+
        #[cfg(test)]
        test_hook_after_manager_start().await;

@@ -0,0 +1,189 @@
+// 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.
+
+//! HS-01 (rustfs/backlog#1865): MRF intent pipeline integration tests.
+//!
+//! Drives the real consumer loop (`spawn_mrf_consumer`) against a real
+//! 4-disk `ECStore` heal storage and a `HealManager` that has not started its
+//! scheduler, so submitted intents stay observable in the admission queue.
+//! Under `cargo nextest` each test runs in its own process, which keeps the
+//! process-global MRF channel singleton safe.
+
+use rustfs_common::mrf_channel::{self, MrfKind};
+use rustfs_heal::heal::{
+    manager::{HealConfig, HealManager},
+    mrf_queue,
+    storage::{ECStoreHealStorage, HealStorageAPI},
+};
+use serial_test::serial;
+use std::{path::Path, sync::Arc, time::Duration};
+
+mod storage_api;
+
+use storage_api::endpoint_index::{Endpoint, EndpointServerPools, Endpoints, PoolEndpoints, init_local_disks};
+
+const META_BUCKET: &str = ".rustfs.sys";
+const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin";
+
+async fn heal_env() -> (Vec<std::path::PathBuf>, Arc<dyn HealStorageAPI>) {
+    let env = rustfs_test_utils::TestECStoreEnv::builder()
+        .prefix("rustfs_heal_mrf_test")
+        .build()
+        .await;
+    let heal_storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
+    (env.disk_paths, heal_storage)
+}
+
+fn make_manager(storage: Arc<dyn HealStorageAPI>) -> Arc<HealManager> {
+    Arc::new(HealManager::new(
+        storage,
+        Some(HealConfig {
+            // Keep the scheduler from draining the queue before assertions.
+            heal_interval: Duration::from_secs(3600),
+            enable_auto_heal: false,
+            ..Default::default()
+        }),
+    ))
+}
+
+/// Encode one journal record independently of the implementation, so a format
+/// drift between writer and this fixture fails loudly here.
+fn journal_record(kind: u8, bucket: &str, object: &str, version: Option<[u8; 16]>, attempts: u8) -> Vec<u8> {
+    let mut body = vec![1u8, 1, kind, attempts];
+    body.extend_from_slice(&1_700_000_000_000u64.to_le_bytes());
+    match version {
+        Some(bytes) => {
+            body.push(1);
+            body.extend_from_slice(&bytes);
+        }
+        None => body.push(0),
+    }
+    body.extend_from_slice(&(bucket.len() as u32).to_le_bytes());
+    body.extend_from_slice(&(object.len() as u32).to_le_bytes());
+    body.extend_from_slice(bucket.as_bytes());
+    body.extend_from_slice(object.as_bytes());
+    let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+    hasher.update(&body);
+    body.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes());
+    body
+}
+
+fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) {
+    for path in disk_paths {
+        let journal = path.join(META_BUCKET).join(JOURNAL_REL);
+        std::fs::create_dir_all(journal.parent().expect("journal parent")).expect("create journal dir");
+        std::fs::write(&journal, data).expect("write journal fixture");
+    }
+}
+
+async fn wait_until<F, Fut>(deadline: Duration, mut probe: F) -> bool
+where
+    F: FnMut() -> Fut,
+    Fut: std::future::Future<Output = bool>,
+{
+    let start = std::time::Instant::now();
+    while start.elapsed() < deadline {
+        if probe().await {
+            return true;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    false
+}
+
+/// A decode-failure intent delivered on the global channel must surface in the
+/// heal manager as an Urgent request attributed to the MRF source.
+#[tokio::test]
+#[serial]
+async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() {
+    let (_disk_paths, storage) = heal_env().await;
+    let manager = make_manager(storage);
+
+    mrf_queue::spawn_mrf_consumer(manager.clone());
+
+    assert!(
+        mrf_channel::try_send_mrf_intent(MrfKind::DecodeFailure, "mrf-bucket", "mrf-object", None),
+        "intent should be accepted while the consumer holds the channel"
+    );
+
+    let appeared = wait_until(Duration::from_secs(10), || async {
+        let snapshot = manager.operations_snapshot().await;
+        snapshot.queued_by_source.mrf >= 1 && snapshot.queued_by_priority.urgent >= 1
+    })
+    .await;
+    assert!(
+        appeared,
+        "MRF intent must reach the manager queue as an Urgent request (snapshot: {:?})",
+        manager.operations_snapshot().await
+    );
+}
+
+/// A journal left behind by a previous process must be replayed into the
+/// manager queue and then removed, and a torn tail must not block replay of
+/// the intact records.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn journal_replay_arms_intents_and_deletes_the_file() {
+    let (disk_paths, storage) = heal_env().await;
+
+    // The journal reader resolves disks through the process-local disk map;
+    // register the environment's disks the same way server startup does.
+    let mut endpoints: Vec<Endpoint> = disk_paths
+        .iter()
+        .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path"))
+        .collect();
+    for (i, endpoint) in endpoints.iter_mut().enumerate() {
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(i);
+    }
+    let pool = PoolEndpoints {
+        legacy: false,
+        set_count: 1,
+        drives_per_set: endpoints.len(),
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: "mrf-test".to_string(),
+        platform: String::new(),
+    };
+    init_local_disks(EndpointServerPools::from(vec![pool]))
+        .await
+        .expect("local disks should register");
+
+    let mut journal = journal_record(1, "replay-bucket", "replay-object", Some([9u8; 16]), 0);
+    journal.extend(journal_record(3, "replay-bucket", "partial-object", None, 1));
+    // Torn tail: a third record truncated mid-way must not block the two
+    // intact records above.
+    journal.extend_from_slice(&journal_record(2, "replay-bucket", "metadata-object", None, 0)[..8]);
+    write_journal_to_disks(&disk_paths, &journal);
+
+    let manager = make_manager(storage);
+    // Replay directly (not via the process-global channel consumer, which the
+    // sibling test already claimed in this process under plain `cargo test`).
+    let replayed = mrf_queue::replay_journal_once(&manager).await;
+    assert_eq!(replayed, 2, "the two intact records must be replayed");
+
+    let snapshot = manager.operations_snapshot().await;
+    assert_eq!(snapshot.queued_by_source.mrf, 2, "replayed intents must be attributed to the MRF source");
+
+    assert!(
+        disk_paths
+            .iter()
+            .all(|path| !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()),
+        "the journal file must be removed after a successful replay"
+    );
+
+    let snapshot = manager.operations_snapshot().await;
+    assert_eq!(snapshot.queued_by_priority.urgent, 1, "the decode-failure record must replay as Urgent");
+    assert!(snapshot.queued_by_priority.normal >= 1, "the partial-write record must replay as Normal");
+}
@@ -26,7 +26,6 @@ pub enum StorageMedia {
 }

 impl StorageMedia {
-    #[allow(dead_code)]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Nvme => "nvme",
@@ -60,7 +59,6 @@ pub enum AccessPattern {
 }

 impl AccessPattern {
-    #[allow(dead_code)]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Sequential => "sequential",
@@ -71,25 +69,21 @@ impl AccessPattern {
    }

    /// Check if this is a sequential access pattern.
-    #[allow(dead_code)]
    pub fn is_sequential(&self) -> bool {
        matches!(self, Self::Sequential)
    }

    /// Check if this is a random access pattern.
-    #[allow(dead_code)]
    pub fn is_random(&self) -> bool {
        matches!(self, Self::Random)
    }

    /// Check if this is a mixed access pattern.
-    #[allow(dead_code)]
    pub fn is_mixed(&self) -> bool {
        matches!(self, Self::Mixed)
    }

    /// Check if this pattern is unknown.
-    #[allow(dead_code)]
    pub fn is_unknown(&self) -> bool {
        matches!(self, Self::Unknown)
    }
@@ -255,6 +255,10 @@ pub use cache::KmsCacheStats;
 pub use config::*;
 pub use deletion_worker::DeletionReferenceChecker;
 pub use encryption::is_data_key_envelope;
+// Re-exported so the object layer binds encryption context exactly the way the
+// KMS backends do. A second canonicalization is how the object layer once
+// serialized a HashMap directly while the Static backend already sorted keys.
+pub use encryption::context_aad;
 pub use error::{KmsError, KmsUnavailableError, Result};
 pub use key_impact::{KeyImpactReport, KeyReference, KeyReferenceKind, ReferenceCompleteness, ReferenceCoverage, ReferenceScope};
 pub use manager::KmsManager;
@@ -427,7 +427,6 @@ pub enum DataSource {
    /// Write triggered
    WriteTriggered,
    /// Fallback value
-    #[allow(dead_code)]
    Fallback,
 }

@@ -603,7 +602,6 @@ impl WriteRecord {

 /// Hybrid strategy configuration
 #[derive(Debug, Clone)]
-#[allow(dead_code)]
 pub struct HybridStrategyConfig {
    /// Scheduled update interval
    pub scheduled_update_interval: Duration,
@@ -998,14 +996,12 @@ impl HybridCapacityManager {
    }

    /// Get cache age
-    #[allow(dead_code)]
    pub async fn get_cache_age(&self) -> Option<Duration> {
        let cache = self.cache.read().await;
        cache.as_ref().map(|c| c.last_update.elapsed())
    }

    /// Get write frequency (writes/minute)
-    #[allow(dead_code)]
    pub async fn get_write_frequency(&self) -> usize {
        let record = &self.write_record;
        record.recent_write_count(record.monotonic_second())
@@ -1300,7 +1296,6 @@ pub fn get_capacity_manager() -> Arc<HybridCapacityManager> {
 ///     .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
 ///     .await;
 /// ```
-#[allow(dead_code)]
 pub fn create_isolated_manager(config: HybridStrategyConfig) -> Arc<HybridCapacityManager> {
    Arc::new(HybridCapacityManager::new(config))
 }
@@ -49,7 +49,6 @@ pub struct IndexInfo {
    pub uncompressed_offset: i64,
 }

-#[allow(dead_code)]
 impl Index {
    pub fn new() -> Self {
        Self {
@@ -60,14 +59,6 @@ impl Index {
        }
    }

-    #[allow(dead_code)]
-    fn reset(&mut self, max_block: usize) {
-        self.est_block_uncomp = max_block as i64;
-        self.total_compressed = -1;
-        self.total_uncompressed = -1;
-        self.info.clear();
-    }
-
    pub fn len(&self) -> usize {
        self.info.len()
    }
@@ -511,47 +502,6 @@ fn read_varint(buf: &[u8]) -> io::Result<(i64, usize)> {
    Err(io::Error::new(io::ErrorKind::UnexpectedEof, "unexpected EOF"))
 }

-// Helper functions for index header manipulation
-#[allow(dead_code)]
-pub fn remove_index_headers(b: &[u8]) -> Option<&[u8]> {
-    if b.len() < 4 + S2_INDEX_TRAILER.len() {
-        return None;
-    }
-
-    // Skip size
-    let b = &b[4..];
-
-    // Check trailer
-    if !b.starts_with(S2_INDEX_TRAILER) {
-        return None;
-    }
-
-    Some(&b[S2_INDEX_TRAILER.len()..])
-}
-
-#[allow(dead_code)]
-pub fn restore_index_headers(in_data: &[u8]) -> Vec<u8> {
-    if in_data.is_empty() {
-        return Vec::new();
-    }
-
-    let mut b = Vec::with_capacity(4 + S2_INDEX_HEADER.len() + in_data.len() + S2_INDEX_TRAILER.len() + 4);
-    b.extend_from_slice(&[0x50, 0x2A, 0x4D, 0x18]);
-    b.extend_from_slice(S2_INDEX_HEADER);
-    b.extend_from_slice(in_data);
-
-    let total_size = (b.len() + 4 + S2_INDEX_TRAILER.len()) as u32;
-    b.extend_from_slice(&total_size.to_le_bytes());
-    b.extend_from_slice(S2_INDEX_TRAILER);
-
-    let chunk_len = b.len() - 4;
-    b[1] = chunk_len as u8;
-    b[2] = (chunk_len >> 8) as u8;
-    b[3] = (chunk_len >> 16) as u8;
-
-    b
-}
-
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -2478,6 +2478,15 @@ impl FolderScanner {
                        }

                        if let GetSizeFailureAction::HealMetadata { object } = failure_action {
+                            // MRF journal intent: durable High-priority Metadata
+                            // heal across restarts (HS-01); the scanner heal
+                            // request below stays as the immediate path.
+                            rustfs_common::mrf_channel::try_send_mrf_intent(
+                                rustfs_common::mrf_channel::MrfKind::MetadataCorruption,
+                                &item.bucket,
+                                &object,
+                                None,
+                            );
                            self.send_required_scanner_heal_request(
                                PendingScannerHealKind::Object,
                                item.bucket.clone(),
@@ -206,7 +206,6 @@ async fn setup_isolated_test_env(init_expiry: bool) -> (Vec<PathBuf>, Arc<ECStor
 }

 /// Test helper: Create a test bucket
-#[allow(dead_code)]
 async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
    (**ecstore)
        .make_bucket(bucket_name, &Default::default())
@@ -251,7 +250,6 @@ async fn modeled_versioned_delete_opts(bucket: &str, object: &str) -> ObjectOpti
 }

 /// Test helper: Set bucket lifecycle configuration
-#[allow(dead_code)]
 async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
    // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing
    let lifecycle_xml = r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -274,7 +272,6 @@ async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::erro
 }

 /// Test helper: Set bucket lifecycle configuration
-#[allow(dead_code)]
 async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
    // Create lifecycle rule that targets delete-marker cleanup only.
    // Keep Expiration.Days unset to avoid expiring live transitioned object versions.
@@ -297,7 +294,6 @@ async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<
    Ok(())
 }

-#[allow(dead_code)]
 async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64) -> Result<(), Box<dyn std::error::Error>> {
    let lifecycle_xml = format!(
        r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -320,7 +316,6 @@ async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64)
    Ok(())
 }

-#[allow(dead_code)]
 async fn set_bucket_lifecycle_transition_with_tier(
    bucket_name: &str,
    storage_class: &str,
@@ -368,7 +363,6 @@ async fn object_exists(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bo
 }

 /// Test helper: Check if object exists
-#[allow(dead_code)]
 async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
    if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
        println!("oi: {oi:?}");
@@ -379,7 +373,6 @@ async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &
    }
 }

-#[allow(dead_code)]
 async fn wait_for_object_absence(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;

@@ -428,7 +428,6 @@ pub fn parse_url(s: &str) -> Result<ParsedURL, NetError> {
    Ok(ParsedURL(uu))
 }

-#[allow(dead_code)]
 pub fn parse_http_url(s: &str) -> Result<ParsedURL, NetError> {
    let u = parse_url(s)?;
    match u.0.scheme() {
@@ -437,7 +436,6 @@ pub fn parse_http_url(s: &str) -> Result<ParsedURL, NetError> {
    }
 }

-#[allow(dead_code)]
 pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> bool {
    if err.kind() == std::io::ErrorKind::TimedOut {
        return !expect_timeouts;
@@ -449,12 +447,10 @@ pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> b
        || err_str.contains("use of closed network connection")
 }

-#[allow(dead_code)]
 pub fn is_conn_reset_err(err: &std::io::Error) -> bool {
    err.to_string().contains("connection reset by peer") || matches!(err.raw_os_error(), Some(libc::ECONNRESET))
 }

-#[allow(dead_code)]
 pub fn is_conn_refused_err(err: &std::io::Error) -> bool {
    err.to_string().contains("connection refused") || matches!(err.raw_os_error(), Some(libc::ECONNREFUSED))
 }
@@ -32,8 +32,8 @@ use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use hyper_rustls::ConfigBuilderExt;
 use rumqttc::{
-    AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, PublishNoticeError, QoS,
-    Transport, mqttbytes::Error as MqttBytesError,
+    AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, ProtocolViolation,
+    PublishNoticeError, PublishOptions, QoS, Transport, mqttbytes::Error as MqttBytesError,
 };
 use rustfs_config::{
    EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST,
@@ -791,7 +791,7 @@ where
                .as_ref()
                .ok_or_else(|| TargetError::Configuration("MQTT client not initialized".to_string()))?;
            let notice = client
-                .publish_tracked(&self.args.topic, self.args.qos, false, body)
+                .publish_tracked(&self.args.topic, body, PublishOptions::new(self.args.qos))
                .await
                .map_err(|error| classify_mqtt_client_error(&error))?;
            drop(client_guard);
@@ -1145,7 +1145,7 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc<Ato
                        );
                        connected_status.store(false, Ordering::SeqCst);
                    }
-                    rumqttc::Event::Incoming(Incoming::PingResp(_)) => {
+                    rumqttc::Event::Incoming(Incoming::PingResp) => {
                        trace!(target_id = %target_id, "Received PingResp from broker. Connection is alive.");
                    }
                    rumqttc::Event::Incoming(Incoming::SubAck(suback)) => {
@@ -1257,7 +1257,11 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc<Ato
 /// copy is preserved and replayed rather than dropped (backlog#971).
 fn classify_mqtt_client_error(err: &ClientError) -> TargetError {
    match err {
-        ClientError::Request(_) | ClientError::TryRequest(_) | ClientError::TrackingUnavailable => TargetError::NotConnected,
+        ClientError::RequestChannelFull(_) | ClientError::RequestChannelDisconnected(_) | ClientError::TrackingUnavailable => {
+            TargetError::NotConnected
+        }
+        ClientError::InvalidRequest(_) => TargetError::Request(format!("Invalid MQTT publish request: {err}")),
+        _ => TargetError::NotConnected,
    }
 }

@@ -1270,10 +1274,14 @@ fn classify_mqtt_notice_error(err: &PublishNoticeError) -> TargetError {
        PublishNoticeError::Recv
        | PublishNoticeError::SessionReset
        | PublishNoticeError::Qos0NotFlushed
+        | PublishNoticeError::BrokerOnlySessionResume
+        | PublishNoticeError::SessionPersistence(_)
        | PublishNoticeError::TopicAliasReplayUnavailable(_) => TargetError::NotConnected,
+        PublishNoticeError::RetainNotSupported => TargetError::Request(format!("MQTT broker rejected publish: {err}")),
        PublishNoticeError::V5PubAck(_) | PublishNoticeError::V5PubRec(_) | PublishNoticeError::V5PubComp(_) => {
            TargetError::Request(format!("MQTT broker rejected publish: {err}"))
        }
+        _ => TargetError::NotConnected,
    }
 }

@@ -1299,12 +1307,13 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool {
                        | MqttBytesError::MalformedPacket // Package format error
                        | MqttBytesError::PayloadTooLong // Too long load
                        | MqttBytesError::PayloadSizeLimitExceeded { .. } // Load size limit exceeded
-                        | MqttBytesError::TopicNotUtf8 // Topic Non-UTF-8 (Serious Agreement Violation)
+                        | MqttBytesError::TopicNotUtf8 { .. } // Topic Non-UTF-8 (Serious Agreement Violation)
                    )
                }
                // Others that are fatal StateError variants
                rumqttc::StateError::InvalidState          // The internal state machine is in invalid state
-                | rumqttc::StateError::WrongPacket         // Agreement Violation: Unexpected Data Packet Received
+                | rumqttc::StateError::ProtocolViolation(ProtocolViolation::UnexpectedIncomingPacket(_)) // Agreement Violation: Unexpected Data Packet Received
+                | rumqttc::StateError::ProtocolViolation(_) // Agreement Violation
                | rumqttc::StateError::Unsolicited(_)      // Agreement Violation: Unsolicited ACK Received
                | rumqttc::StateError::CollisionTimeout    // Agreement Violation (if this stage occurs)
                | rumqttc::StateError::EmptySubscription   // Agreement violation (if this stage occurs)
@@ -1727,8 +1736,8 @@ where
 mod tests {
    use super::{
        AsyncClient, ClientError, MQTT_RECONNECT_BACKOFF_MAX, MQTT_RECONNECT_BACKOFF_MIN, MQTTArgs, MQTTTarget, MQTTTlsConfig,
-        MqttOptions, PublishNoticeError, QoS, QueuedPayloadMeta, classify_mqtt_client_error, classify_mqtt_notice_error,
-        next_reconnect_backoff, reconnect_supervisor, validate_mqtt_broker_url,
+        MqttOptions, PublishNoticeError, PublishOptions, QoS, QueuedPayloadMeta, classify_mqtt_client_error,
+        classify_mqtt_notice_error, next_reconnect_backoff, reconnect_supervisor, validate_mqtt_broker_url,
    };
    use crate::error::TargetError;
    use crate::target::{REDACTED_SECRET, TargetType};
@@ -1794,7 +1803,7 @@ mod tests {
            .capacity(1)
            .build();
        client
-            .publish("fill", QoS::AtLeastOnce, false, b"fill".as_slice())
+            .publish("fill", b"fill".as_slice(), PublishOptions::new(QoS::AtLeastOnce))
            .await
            .expect("first publish should fill the local channel");
        *target.client.lock().await = Some(client);
@@ -107,7 +107,7 @@ inventory. Generic function-local names such as `CACHE`, `LOCK`, `INIT`, and
 | `AUTH_FS` | `rustfs/src/storage/access.rs` | Cache or constant / owner-local cache | Authorization tag-condition lookup keeps its filesystem helper private to the access owner. |
 | `LOCK_STATS` | `rustfs/src/storage/lock_optimizer.rs` | Process-global owner-local metrics | Lock optimization statistics stay private behind lock optimizer helper APIs. |
 | `DEADLOCK_DETECTOR` | `rustfs/src/storage/deadlock_detector.rs` | Process-global owner-local state | Deadlock detector lifecycle state stays private to the storage deadlock detector owner. |
-| `CONCURRENCY_MANAGER`, `ACTIVE_GET_REQUESTS`, `ACTIVE_PUT_REQUESTS`, `IO_PRIORITY_METRICS` | `rustfs/src/storage/concurrency/*` | Process-global owner-local scheduler state | Storage concurrency manager, counters, and metrics remain inside the storage concurrency owner boundary. |
+| `CONCURRENCY_MANAGER`, `ACTIVE_GET_REQUESTS`, `ACTIVE_PUT_REQUESTS` | `rustfs/src/storage/concurrency/*` | Process-global owner-local scheduler state | Storage concurrency manager and request counters remain inside the storage concurrency owner boundary. |
 | `GET_OBJECT_BUFFER_THRESHOLD_WARNED`, `GET_READER_STREAM_BUFFER_SIZE_OVERRIDE`, function-local `ENABLED`, `OBJECT_SEEK_SUPPORT_THRESHOLD`, `OBJECT_SEEK_SUPPORT_CONCURRENCY_THRESHOLDS` | `rustfs/src/app/object_usecase.rs` | Cache or constant / owner-local cache | Object GET/seek tuning caches and warning guards stay private to object usecase helpers. |
 | `SUPPORTED_HEADERS` | `rustfs/src/storage/options.rs` | Cache or constant / owner-local constant | Supported-header lookup state stays private to storage option parsing. |
 | `AUDIT_TARGET_SPECS`, `NOTIFICATION_TARGET_SPECS` | `rustfs/src/admin/handlers/audit.rs`, `rustfs/src/admin/handlers/event.rs`, `rustfs/src/admin/handlers/plugins_instances.rs` | Cache or constant / owner-local constant | Admin target descriptor tables stay private to their handler owners. |
@@ -317,6 +317,7 @@ fn add_source_counts(total: &mut rustfs_heal::HealSourceCounts, next: rustfs_hea
    total.auto_heal = total.auto_heal.saturating_add(next.auto_heal);
    total.internal = total.internal.saturating_add(next.internal);
    total.read_repair = total.read_repair.saturating_add(next.read_repair);
+    total.mrf = total.mrf.saturating_add(next.mrf);
 }

 fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_heal::HealOperationsSnapshot) {
@@ -2353,6 +2354,7 @@ mod tests {
            auto_heal: value,
            internal: value,
            read_repair: value,
+            mrf: value,
        };
        let operations = |value| rustfs_heal::HealOperationsSnapshot {
            queue_length: value,
@@ -16,13 +16,12 @@

 use super::kms_dynamic::current_kms_config_fingerprint;
 use super::kms_keys::{CreateKeyHandler, DescribeKeyHandler, GenerateDataKeyHandler, ListKeysHandler};
-use crate::admin::auth::validate_admin_request;
+use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{
    current_kms_runtime_service_manager, current_notification_system, current_or_init_kms_runtime_service_manager,
 };
-use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::ADMIN_PREFIX;
 use hyper::{HeaderMap, Method, StatusCode};
 use matchit::Params;
 use rustfs_kms::KmsBackend;
@@ -69,6 +68,18 @@ fn kms_clear_cache_actions() -> Vec<Action> {
    vec![Action::KmsAction(KmsAction::ClearCacheAction)]
 }

+/// Admin gate for the KMS management endpoints, none of which act on a key.
+///
+/// The pre-check keeps these endpoints' historical missing-credentials message;
+/// the shared gate reports "get cred failed".
+async fn authorize_kms_management_request(req: &S3Request<Body>, actions: Vec<Action>) -> S3Result<()> {
+    if req.credentials.is_none() {
+        return Err(s3_error!(InvalidRequest, "authentication required"));
+    }
+    authorize_admin_request(req, actions).await?;
+    Ok(())
+}
+
 /// Response of `POST /kms/clear-cache`.
 ///
 /// Declared rather than built inline so the shape the console already depends
@@ -260,22 +271,7 @@ pub struct KmsStatusHandler {}
 #[async_trait::async_trait]
 impl Operation for KmsStatusHandler {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let Some(cred) = req.credentials else {
-            return Err(s3_error!(InvalidRequest, "authentication required"));
-        };
-
-        let (cred, owner) =
-            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
-
-        validate_admin_request(
-            &req.headers,
-            &cred,
-            owner,
-            false,
-            kms_service_control_actions(),
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
-        )
-        .await?;
+        authorize_kms_management_request(&req, kms_service_control_actions()).await?;

        let Some(service) = kms_encryption_service_from_context().await else {
            return Err(s3_error!(InternalError, "KMS service not initialized"));
@@ -326,22 +322,7 @@ pub struct KmsConfigHandler {}
 #[async_trait::async_trait]
 impl Operation for KmsConfigHandler {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let Some(cred) = req.credentials else {
-            return Err(s3_error!(InvalidRequest, "authentication required"));
-        };
-
-        let (cred, owner) =
-            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
-
-        validate_admin_request(
-            &req.headers,
-            &cred,
-            owner,
-            false,
-            kms_configure_actions(),
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
-        )
-        .await?;
+        authorize_kms_management_request(&req, kms_configure_actions()).await?;

        let Some(service) = kms_encryption_service_from_context().await else {
            return Err(s3_error!(InternalError, "KMS service not initialized"));
@@ -375,22 +356,7 @@ pub struct KmsClearCacheHandler {}
 #[async_trait::async_trait]
 impl Operation for KmsClearCacheHandler {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let Some(cred) = req.credentials else {
-            return Err(s3_error!(InvalidRequest, "authentication required"));
-        };
-
-        let (cred, owner) =
-            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
-
-        validate_admin_request(
-            &req.headers,
-            &cred,
-            owner,
-            false,
-            kms_clear_cache_actions(),
-            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
-        )
-        .await?;
+        authorize_kms_management_request(&req, kms_clear_cache_actions()).await?;

        let Some(service) = kms_encryption_service_from_context().await else {
            return Err(s3_error!(InternalError, "KMS service not initialized"));
@@ -422,9 +388,14 @@ impl Operation for KmsClearCacheHandler {

 #[cfg(test)]
 mod tests {
-    use super::{KmsClearCacheResponse, kms_clear_cache_actions, kms_configure_actions, kms_service_control_actions};
+    use super::{
+        KmsClearCacheResponse, authorize_kms_management_request, kms_clear_cache_actions, kms_configure_actions,
+        kms_service_control_actions,
+    };
    use crate::admin::handlers::kms_keys::stable_json_value;
+    use hyper::HeaderMap;
    use rustfs_policy::policy::action::{Action, AdminAction, KmsAction};
+    use s3s::{Body, S3Request};

    fn assert_has_action(actions: &[Action], action: Action) {
        assert!(actions.contains(&action), "expected action list to contain {action:?}");
@@ -434,6 +405,58 @@ mod tests {
        assert!(!actions.contains(&action), "expected action list not to contain {action:?}");
    }

+    /// These endpoints authorize through the shared admin gate, which reports
+    /// "get cred failed" for a credential-less request. The pre-check keeps the
+    /// message these endpoints have always returned (rustfs/backlog#1829).
+    #[tokio::test]
+    async fn kms_management_gate_keeps_its_missing_credentials_message() {
+        let req = S3Request {
+            input: Body::from(String::new()),
+            method: http::Method::GET,
+            uri: "/rustfs/admin/v3/kms/status".parse().expect("uri should parse"),
+            headers: HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let err = authorize_kms_management_request(&req, kms_service_control_actions())
+            .await
+            .expect_err("a request without credentials must be rejected");
+        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
+        assert_eq!(err.message(), Some("authentication required"));
+    }
+
+    /// Every management endpoint must reach the shared gate, each with its own
+    /// action set. The action lists are pinned above, but nothing else checks
+    /// which handler asks for which, and a handler that lost its gate entirely
+    /// would still serve its response.
+    #[test]
+    fn management_handlers_authorize_with_their_dedicated_actions() {
+        let src = include_str!("kms_management.rs");
+
+        for (handler, actions) in [
+            ("KmsStatusHandler", "kms_service_control_actions()"),
+            ("KmsConfigHandler", "kms_configure_actions()"),
+            ("KmsClearCacheHandler", "kms_clear_cache_actions()"),
+        ] {
+            let block = src
+                .split_once(&format!("impl Operation for {handler}"))
+                .unwrap_or_else(|| panic!("{handler} impl should exist"))
+                .1;
+            let end = block
+                .find("\nimpl Operation for")
+                .or_else(|| block.find("\n#[cfg(test)]"))
+                .unwrap_or(block.len());
+            assert!(
+                block[..end].contains(&format!("authorize_kms_management_request(&req, {actions})")),
+                "{handler} must authorize through the shared gate with {actions}"
+            );
+        }
+    }
+
    #[test]
    fn kms_management_auth_actions_use_dedicated_kms_actions() {
        assert_has_action(&kms_service_control_actions(), Action::KmsAction(KmsAction::ServiceControlAction));
@@ -56,8 +56,8 @@ use super::storage_api::object_usecase::bucket::{
 };
 use super::storage_api::object_usecase::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
 use super::storage_api::object_usecase::concurrency::{
-    self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size,
-    get_concurrency_manager, get_put_concurrency_aware_buffer_size,
+    self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectAdmission, PutObjectGuard,
+    get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
 };
 #[cfg(test)]
 use super::storage_api::object_usecase::contract::http::HTTPPreconditions;
@@ -5681,6 +5681,35 @@ impl DefaultObjectUsecase {
        let server_side_encryption_requested =
            server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some();

+        // Resolve the store through the request-bound server context
+        // (backlog#1052 S6), not the process-global handle, so an embedded
+        // second server never writes into the first server's store.
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now);
+        validate_bucket_exists(&store, &bucket).await?;
+        rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start);
+
+        let put_admission = match get_concurrency_manager()
+            .admit_put_object()
+            .await
+            .map_err(|_| s3_error!(InternalError, "foreground write admission closed"))?
+        {
+            PutObjectAdmission::Disabled => None,
+            PutObjectAdmission::Admitted(permit) => {
+                counter!("rustfs.put_object.foreground_admission.total", "result" => "admitted").increment(1);
+                Some(permit)
+            }
+            PutObjectAdmission::Rejected => {
+                counter!("rustfs.put_object.foreground_admission.total", "result" => "rejected").increment(1);
+                return Err(s3_error!(
+                    SlowDown,
+                    "foreground write concurrency limit reached, please reduce your request rate"
+                ));
+            }
+        };
+
        let mut put_request_guard = PutObjectGuard::new();
        let concurrent_put_requests = PutObjectGuard::concurrent_requests();

@@ -5733,16 +5762,6 @@ impl DefaultObjectUsecase {
            use_large_put_concurrency_tuning,
        );

-        // Resolve the store through the request-bound server context
-        // (backlog#1052 S6), not the process-global handle, so an embedded
-        // second server never writes into the first server's store.
-        let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
-        let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        validate_bucket_exists(&store, &bucket).await?;
-        rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start);
-
        let sse_config_stage_start = put_stage_metrics_enabled.then(Instant::now);
        let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
        rustfs_io_metrics::record_put_object_stage_duration_from("app_sse_config_lookup", sse_config_stage_start);
@@ -6132,7 +6151,9 @@ impl DefaultObjectUsecase {
            let cache_adapter = cache_adapter.clone();
            let request_id = request_id.clone();
            let put_path = put_path.to_string();
+            let put_admission = put_admission;
            async move {
+                let _put_admission = put_admission;
                let object_traffic_progress = object_traffic_health
                    .as_deref()
                    .and_then(ObjectTrafficHealth::track_write_storage);
@@ -6183,6 +6204,7 @@ impl DefaultObjectUsecase {
                    }
                };
                rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start);
+                drop(_put_admission);
                drop(object_traffic_progress);
                #[cfg(test)]
                wait_for_put_post_store_test_hook(&bucket).await;
@@ -936,7 +936,7 @@ pub(crate) mod bucket {

 pub(crate) mod concurrency {
    pub(crate) use crate::storage::storage_api::concurrency_consumer::{
-        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard,
+        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectAdmission, PutObjectGuard,
        get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
    };
 }
@@ -14,21 +14,12 @@

 //! I/O scheduling types for adaptive buffer sizing and load management.
 //!
-//! # Migration Note
-//!
-//! This module contains types that are also available in `rustfs_io_core`.
-//! For new code, prefer using types from `rustfs_io_core` directly:
-//!
-//! ```ignore
-//! // Recommended: Use io-core types
-//! use rustfs_io_core::{
-//!     IoLoadLevel, IoPriority, IoSchedulerConfig,
-//!     calculate_optimal_buffer_size, get_buffer_size_for_media,
-//! };
-//! ```
-//!
-//! This module remains for backward compatibility and provides additional
-//! runtime monitoring features (`IoPriorityMetrics`, `IoStrategyDebugInfo`).
+//! This is the live scheduling implementation. `rustfs_io_core` supplies the
+//! shared config shapes (`IoSchedulerConfig`, `IoPriorityQueueConfig`) that the
+//! types here project into through `to_core_config`, plus the `io_profile`
+//! storage-media model; bandwidth samples come from `rustfs_io_metrics`.
+//! Same-named io-core types are those config shapes, not a backing
+//! implementation this module delegates to.

 use rustfs_config::{KI_B, MI_B};
 use rustfs_io_core::io_profile::{AccessPattern, StorageMedia, StorageProfile};
@@ -1762,169 +1753,6 @@ impl<T> IoPriorityQueue<T> {
    }
 }

-// ============================================
-// I/O Priority Queue Metrics
-// ============================================
-
-/// Global metrics for I/O priority queue monitoring.
-///
-/// These metrics are emitted through the shared metrics pipeline and provide
-/// visibility into the priority queue behavior.
-#[allow(dead_code)]
-pub struct IoPriorityMetrics {
-    /// High priority queue depth.
-    pub high_queue_depth: AtomicU64,
-    /// Normal priority queue depth.
-    pub normal_queue_depth: AtomicU64,
-    /// Low priority queue depth.
-    pub low_queue_depth: AtomicU64,
-    /// High priority total wait time in nanoseconds.
-    pub high_wait_time_ns: AtomicU64,
-    /// Normal priority total wait time in nanoseconds.
-    pub normal_wait_time_ns: AtomicU64,
-    /// Low priority total wait time in nanoseconds.
-    pub low_wait_time_ns: AtomicU64,
-    /// Total starvation events count.
-    pub starvation_events: AtomicU64,
-    /// High priority requests processed.
-    pub high_processed: AtomicU64,
-    /// Normal priority requests processed.
-    pub normal_processed: AtomicU64,
-    /// Low priority requests processed.
-    pub low_processed: AtomicU64,
-}
-
-#[allow(dead_code)]
-impl Default for IoPriorityMetrics {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-#[allow(dead_code)]
-impl IoPriorityMetrics {
-    /// Create a new metrics instance.
-    pub const fn new() -> Self {
-        Self {
-            high_queue_depth: AtomicU64::new(0),
-            normal_queue_depth: AtomicU64::new(0),
-            low_queue_depth: AtomicU64::new(0),
-            high_wait_time_ns: AtomicU64::new(0),
-            normal_wait_time_ns: AtomicU64::new(0),
-            low_wait_time_ns: AtomicU64::new(0),
-            starvation_events: AtomicU64::new(0),
-            high_processed: AtomicU64::new(0),
-            normal_processed: AtomicU64::new(0),
-            low_processed: AtomicU64::new(0),
-        }
-    }
-
-    /// Update queue depths from status.
-    #[allow(dead_code)]
-    pub fn update_queue_depths(&self, status: &IoQueueStatus) {
-        self.high_queue_depth
-            .store(status.high_priority_waiting as u64, Ordering::Relaxed);
-        self.normal_queue_depth
-            .store(status.normal_priority_waiting as u64, Ordering::Relaxed);
-        self.low_queue_depth
-            .store(status.low_priority_waiting as u64, Ordering::Relaxed);
-    }
-
-    /// Record a starvation event.
-    #[allow(dead_code)]
-    pub fn record_starvation(&self) {
-        self.starvation_events.fetch_add(1, Ordering::Relaxed);
-    }
-
-    /// Record a processed request.
-    #[allow(dead_code)]
-    pub fn record_processed(&self, priority: IoPriority) {
-        match priority {
-            IoPriority::High => self.high_processed.fetch_add(1, Ordering::Relaxed),
-            IoPriority::Normal => self.normal_processed.fetch_add(1, Ordering::Relaxed),
-            IoPriority::Low => self.low_processed.fetch_add(1, Ordering::Relaxed),
-        };
-    }
-
-    /// Record wait time for a priority level.
-    pub fn record_wait_time(&self, priority: IoPriority, wait_ns: u64) {
-        match priority {
-            IoPriority::High => self.high_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed),
-            IoPriority::Normal => self.normal_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed),
-            IoPriority::Low => self.low_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed),
-        };
-    }
-
-    /// Get high priority queue depth.
-    pub fn get_high_queue_depth(&self) -> u64 {
-        self.high_queue_depth.load(Ordering::Relaxed)
-    }
-
-    /// Get normal priority queue depth.
-    pub fn get_normal_queue_depth(&self) -> u64 {
-        self.normal_queue_depth.load(Ordering::Relaxed)
-    }
-
-    /// Get low priority queue depth.
-    pub fn get_low_queue_depth(&self) -> u64 {
-        self.low_queue_depth.load(Ordering::Relaxed)
-    }
-
-    /// Get total starvation events.
-    pub fn get_starvation_events(&self) -> u64 {
-        self.starvation_events.load(Ordering::Relaxed)
-    }
-
-    /// Get metrics summary for logging/debugging.
-    pub fn summary(&self) -> String {
-        format!(
-            "high_queue={}, normal_queue={}, low_queue={}, starvation={}, high_proc={}, normal_proc={}, low_proc={}",
-            self.get_high_queue_depth(),
-            self.get_normal_queue_depth(),
-            self.get_low_queue_depth(),
-            self.get_starvation_events(),
-            self.high_processed.load(Ordering::Relaxed),
-            self.normal_processed.load(Ordering::Relaxed),
-            self.low_processed.load(Ordering::Relaxed)
-        )
-    }
-}
-
-/// Global I/O priority metrics instance.
-#[allow(dead_code)]
-pub static IO_PRIORITY_METRICS: IoPriorityMetrics = IoPriorityMetrics::new();
-
-/// Get optimized buffer size for I/O operations.
-///
-/// This function provides adaptive buffer sizing based on:
-/// - File size (small files get smaller buffers)
-/// - Concurrent request count (high concurrency gets smaller buffers)
-/// - Base buffer size from configuration
-///
-/// # Arguments
-///
-/// * `file_size` - Size of the file being read/written (-1 for unknown)
-///
-/// # Returns
-///
-/// Optimal buffer size in bytes
-///
-/// # Example
-///
-/// ```ignore
-/// let buffer_size = get_buffer_size_opt_in(1024 * 1024); // 1MB file
-/// assert!(buffer_size >= 64 * 1024); // At least 64KB
-/// ```
-#[allow(dead_code)]
-pub fn get_buffer_size_opt_in(file_size: i64) -> usize {
-    // Get base buffer size from configuration
-    let base_buffer_size =
-        rustfs_utils::get_env_usize(rustfs_config::ENV_OBJECT_IO_BUFFER_SIZE, rustfs_config::DEFAULT_OBJECT_IO_BUFFER_SIZE);
-
-    // Apply concurrency-aware adjustments
-    get_concurrency_aware_buffer_size(file_size, base_buffer_size)
-}
-
 // ============================================
 // Unit Tests
 // ============================================
@@ -1933,13 +1761,12 @@ pub fn get_buffer_size_opt_in(file_size: i64) -> usize {
 #[allow(unused_imports)]
 mod tests {
    use super::{
-        IoLoadLevel, IoPriority, IoPriorityMetrics, IoPriorityQueue, IoPriorityQueueConfig, IoSchedulerConfig,
-        IoSchedulingContext, IoStrategy, get_advanced_buffer_size, get_buffer_size_opt_in, get_concurrency_aware_buffer_size,
+        IoLoadLevel, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoSchedulerConfig, IoSchedulingContext, IoStrategy,
+        get_advanced_buffer_size, get_concurrency_aware_buffer_size,
    };
    use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
    use rustfs_io_metrics::bandwidth::{BandwidthSnapshot, BandwidthTier};
    use serial_test::serial;
-    use std::sync::atomic::Ordering;
    use std::time::Duration;

    #[tokio::test]
@@ -2126,29 +1953,6 @@ mod tests {
        assert_eq!(config.starvation_threshold_secs, 120);
    }

-    #[tokio::test]
-    #[serial]
-    async fn test_io_priority_metrics() {
-        let metrics = IoPriorityMetrics::new();
-
-        // Test initial state
-        assert_eq!(metrics.get_high_queue_depth(), 0);
-        assert_eq!(metrics.get_normal_queue_depth(), 0);
-        assert_eq!(metrics.get_low_queue_depth(), 0);
-        assert_eq!(metrics.get_starvation_events(), 0);
-
-        // Test recording
-        metrics.record_starvation();
-        assert_eq!(metrics.get_starvation_events(), 1);
-
-        metrics.record_processed(IoPriority::High);
-        metrics.record_processed(IoPriority::High);
-        metrics.record_processed(IoPriority::Normal);
-
-        assert_eq!(metrics.high_processed.load(Ordering::Relaxed), 2);
-        assert_eq!(metrics.normal_processed.load(Ordering::Relaxed), 1);
-    }
-
    // ============================================
    // Multi-Factor Strategy Tests
    // ============================================
@@ -65,6 +65,11 @@ pub struct ConcurrencyManager {
    bandwidth_monitor: Arc<Mutex<BandwidthMonitor>>,
    /// Metrics collector for I/O latency tracking (P50, P95, P99)
    metrics_collector: Arc<MetricsCollector>,
+    /// Experimental fixed-count foreground PutObject admission gate.
+    put_admission_semaphore: Arc<Semaphore>,
+    put_admission_enabled: bool,
+    put_admission_limit: usize,
+    put_admission_wait_timeout: Duration,
 }

 impl std::fmt::Debug for ConcurrencyManager {
@@ -114,6 +119,18 @@ pub enum DiskReadAdmission {
    Rejected,
 }

+/// Outcome of foreground PutObject request admission.
+#[derive(Debug)]
+pub enum PutObjectAdmission {
+    /// Foreground PUT admission is disabled; proceed on the legacy path.
+    Disabled,
+    /// Request is admitted and must hold the permit until the store write
+    /// returns or the request fails before mutation.
+    Admitted(tokio::sync::OwnedSemaphorePermit),
+    /// The fixed-count gate stayed full until the configured wait timeout.
+    Rejected,
+}
+
 #[allow(dead_code)]
 impl ConcurrencyManager {
    /// Create a new concurrency manager with default settings
@@ -161,6 +178,18 @@ impl ConcurrencyManager {
        // Initialize metrics collector for I/O latency tracking
        // Keep 1000 samples for P95/P99 calculation
        let metrics_collector = Arc::new(MetricsCollector::new(performance_metrics, 1000));
+        let put_admission_enabled = rustfs_utils::get_env_bool(
+            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_ENABLE,
+            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE,
+        );
+        let put_admission_limit = rustfs_utils::get_env_usize(
+            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_LIMIT,
+            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT,
+        );
+        let put_admission_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64(
+            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
+            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
+        ));

        // Build queue config directly from scheduler config.
        let queue_config = IoPriorityQueueConfig::from_scheduler_config(&scheduler_config);
@@ -176,6 +205,10 @@ impl ConcurrencyManager {
            pattern_detector,
            bandwidth_monitor,
            metrics_collector,
+            put_admission_semaphore: Arc::new(Semaphore::new(if put_admission_enabled { put_admission_limit } else { 0 })),
+            put_admission_enabled,
+            put_admission_limit,
+            put_admission_wait_timeout,
        }
    }

@@ -199,6 +232,16 @@ impl ConcurrencyManager {
        self.degraded_read_semaphore.close();
    }

+    #[cfg(test)]
+    pub(crate) fn with_put_admission_for_test(enabled: bool, limit: usize, wait_timeout: Duration) -> Self {
+        let mut manager = Self::new();
+        manager.put_admission_semaphore = Arc::new(Semaphore::new(if enabled { limit } else { 0 }));
+        manager.put_admission_enabled = enabled;
+        manager.put_admission_limit = limit;
+        manager.put_admission_wait_timeout = wait_timeout;
+        manager
+    }
+
    /// Track a GetObject request
    pub fn track_request() -> GetObjectGuard {
        GetObjectGuard::new()
@@ -284,6 +327,32 @@ impl ConcurrencyManager {
        }
    }

+    /// Admit a foreground PutObject request under the experimental fixed-count gate.
+    ///
+    /// The default-off path returns [`PutObjectAdmission::Disabled`] without
+    /// touching the semaphore, preserving legacy behavior. When enabled, the
+    /// permit must be acquired before body ingest and held until the store write
+    /// returns, so saturated foreground writes can fail with `SlowDown` before
+    /// creating visible side effects.
+    pub async fn admit_put_object(&self) -> Result<PutObjectAdmission, tokio::sync::AcquireError> {
+        if !self.put_admission_enabled || self.put_admission_limit == 0 {
+            return Ok(PutObjectAdmission::Disabled);
+        }
+
+        if self.put_admission_wait_timeout.is_zero() {
+            return Ok(match self.put_admission_semaphore.clone().try_acquire_owned() {
+                Ok(permit) => PutObjectAdmission::Admitted(permit),
+                Err(tokio::sync::TryAcquireError::NoPermits) => PutObjectAdmission::Rejected,
+                Err(tokio::sync::TryAcquireError::Closed) => PutObjectAdmission::Rejected,
+            });
+        }
+
+        match tokio::time::timeout(self.put_admission_wait_timeout, self.put_admission_semaphore.clone().acquire_owned()).await {
+            Ok(permit) => Ok(PutObjectAdmission::Admitted(permit?)),
+            Err(_) => Ok(PutObjectAdmission::Rejected),
+        }
+    }
+
    // ============================================
    // Adaptive I/O Strategy Methods
    // ============================================
@@ -692,8 +761,16 @@ impl ConcurrencyManager {

    /// Get a read-only workload admission snapshot for foreground writes.
    pub fn put_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot {
-        let active = PutObjectGuard::concurrent_count();
-        let limit = self.scheduler_config.max_concurrent_reads;
+        let (active, limit, hard_gate_enabled) = if self.put_admission_enabled && self.put_admission_limit > 0 {
+            (
+                self.put_admission_limit
+                    .saturating_sub(self.put_admission_semaphore.available_permits()),
+                self.put_admission_limit,
+                true,
+            )
+        } else {
+            (PutObjectGuard::concurrent_count(), self.scheduler_config.max_concurrent_reads, false)
+        };
        let state = if limit == 0 {
            AdmissionState::Disabled
        } else if active >= limit {
@@ -706,7 +783,10 @@ impl ConcurrencyManager {
            WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit));

        match state {
-            AdmissionState::Disabled => admission.with_reason("foreground write pressure tracking disabled"),
+            AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"),
+            AdmissionState::Saturated if hard_gate_enabled => {
+                admission.with_reason("foreground write admission permits exhausted")
+            }
            AdmissionState::Saturated => admission.with_reason("foreground write concurrency reached local pressure limit"),
            _ => admission,
        }
@@ -783,7 +863,7 @@ impl Default for ConcurrencyManager {
 mod integration_tests {
    use super::super::io_schedule::{IoLoadLevel, IoPriority};
    use super::super::request_guard::GetObjectGuard;
-    use super::ConcurrencyManager;
+    use super::{ConcurrencyManager, PutObjectAdmission};
    use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
    use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
    use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
@@ -880,6 +960,71 @@ mod integration_tests {
        crate::storage::concurrency::reset_active_put_requests();
    }

+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_put_admission_disabled_does_not_touch_gate() {
+        let manager = ConcurrencyManager::with_put_admission_for_test(false, 1, Duration::ZERO);
+
+        let admission = manager
+            .admit_put_object()
+            .await
+            .expect("disabled put admission must not close");
+
+        assert!(matches!(admission, PutObjectAdmission::Disabled));
+        assert_eq!(manager.put_admission_semaphore.available_permits(), 0);
+        assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Open);
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_put_admission_rejects_when_limit_full() {
+        let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO);
+
+        let first = manager.admit_put_object().await.expect("first put admission should acquire");
+        assert!(matches!(first, PutObjectAdmission::Admitted(_)));
+        assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Saturated);
+
+        let second = manager
+            .admit_put_object()
+            .await
+            .expect("full put admission gate should reject, not close");
+        assert!(matches!(second, PutObjectAdmission::Rejected));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_put_admission_reuses_released_permit() {
+        let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO);
+
+        let first = manager.admit_put_object().await.expect("first put admission should acquire");
+        drop(first);
+
+        let second = manager
+            .admit_put_object()
+            .await
+            .expect("released put admission permit should be reusable");
+        assert!(matches!(second, PutObjectAdmission::Admitted(_)));
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial]
+    async fn test_concurrency_manager_put_admission_wait_timeout_rejects() {
+        let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::from_secs(5));
+        let held = manager.admit_put_object().await.expect("first put admission should acquire");
+        let waiter_manager = manager.clone();
+
+        let waiter = tokio::spawn(async move { waiter_manager.admit_put_object().await });
+        tokio::task::yield_now().await;
+        tokio::time::advance(Duration::from_secs(5)).await;
+
+        let admission = waiter
+            .await
+            .expect("put admission waiter task must not panic")
+            .expect("put admission gate must stay open");
+        assert!(matches!(admission, PutObjectAdmission::Rejected));
+        drop(held);
+    }
+
    #[tokio::test]
    #[serial]
    async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() {
@@ -24,16 +24,14 @@
 //! - **Concurrency Management**: Coordination of concurrent GetObject requests
 //! - **Request Tracking**: RAII guards for request lifecycle management
 //!
-//! # Migration Note
+//! # Relationship to the shared crates
 //!
-//! Core algorithms have been migrated to `rustfs-io-core` and metrics to
-//! `rustfs-io-metrics`. This module maintains API compatibility while
-//! delegating to the new implementations.
+//! The scheduling algorithm lives in [`io_schedule`], not in `rustfs-io-core`:
+//! this module does not delegate to it. `rustfs-io-core` owns the shared
+//! config shapes and the `io_profile` storage-media model that [`io_schedule`]
+//! consumes, and `rustfs-io-metrics` owns bandwidth sampling and metric
+//! recording.

-// Sub-modules
-// pub mod bandwidth_monitor; // Migrated to rustfs-io-metrics
-// pub mod global_metrics; // Migrated to rustfs-io-metrics
-// pub mod io_profile; // Migrated to rustfs-io-core
 pub mod io_schedule;
 pub mod manager;
 pub mod request_guard;
@@ -45,34 +43,15 @@ pub mod request_guard;
 // I/O scheduling types (from io_schedule.rs for backward compatibility)
 #[allow(unused_imports)]
 pub use io_schedule::{
-    IO_PRIORITY_METRICS, IoLoadLevel, IoPriority, IoPriorityMetrics, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus,
-    IoSchedulerConfig, IoStrategy, get_advanced_buffer_size, get_buffer_size_opt_in, get_concurrency_aware_buffer_size,
-    get_put_concurrency_aware_buffer_size,
+    IoLoadLevel, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, IoSchedulerConfig, IoStrategy,
+    get_advanced_buffer_size, get_concurrency_aware_buffer_size, get_put_concurrency_aware_buffer_size,
 };

 // Request tracking
 pub use request_guard::{GetObjectGuard, PutObjectGuard};

 // Concurrency manager
-pub use manager::{ConcurrencyManager, DiskReadAdmission};
-
-// ============================================
-// New Module Re-exports (for gradual migration)
-// ============================================
-
-// Re-export types from rustfs-io-core for convenience
-pub use rustfs_io_core::{
-    // Backpressure types
-    BackpressureMonitor,
-    // Deadlock detection types
-    DeadlockDetector,
-    // Scheduler types
-    IoScheduler,
-    // Lock optimization types
-    LockOptimizer,
-};
-
-// Re-export types from rustfs-io-metrics for convenience
+pub use manager::{ConcurrencyManager, DiskReadAdmission, PutObjectAdmission};

 // ============================================
 // Helper Functions
@@ -83,37 +62,8 @@ pub fn get_concurrency_manager() -> &'static ConcurrencyManager {
    ConcurrencyManager::global()
 }

-/// Reset the active get requests counter (for testing).
-#[allow(dead_code)]
-pub fn reset_active_get_requests() {
-    io_schedule::ACTIVE_GET_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed);
-}
-
-#[allow(dead_code)]
+/// Reset the active put requests counter (for testing).
+#[cfg(test)]
 pub fn reset_active_put_requests() {
    io_schedule::ACTIVE_PUT_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed);
 }
-
-/// Create a new I/O scheduler with default configuration.
-#[allow(dead_code)]
-pub fn create_io_scheduler() -> IoScheduler {
-    IoScheduler::with_defaults()
-}
-
-/// Create a new backpressure monitor with default configuration.
-#[allow(dead_code)]
-pub fn create_backpressure_monitor() -> BackpressureMonitor {
-    BackpressureMonitor::with_defaults()
-}
-
-/// Create a new deadlock detector with default configuration.
-#[allow(dead_code)]
-pub fn create_deadlock_detector() -> DeadlockDetector {
-    DeadlockDetector::with_defaults()
-}
-
-/// Create a new lock optimizer with default configuration.
-#[allow(dead_code)]
-pub fn create_lock_optimizer() -> LockOptimizer {
-    LockOptimizer::with_defaults()
-}
@@ -4,7 +4,7 @@ use std::fs;
 use std::io::Cursor;
 use std::path::{Path, PathBuf};

-use super::sse::SseObjectEncryptionResolver;
+use super::sse::{SseObjectEncryptionResolver, reset_sse_dek_provider};
 use super::storage_api::ecstore_test_support::{
    DiskAPI as _, DiskOption, Endpoint, Erasure, GetObjectReader, ObjectInfo, ObjectOptions, create_bitrot_reader, new_disk,
 };
@@ -131,6 +131,13 @@ async fn load_fixture_reader_input(case_id: &str) -> (ObjectInfo, Vec<u8>, Strin
 async fn read_fixture_plaintext(encrypted: Vec<u8>, object_info: ObjectInfo, kms_key_b64: String) -> Result<Vec<u8>, String> {
    let object_size = object_info.size;

+    // The DEK provider is cached process-wide once built, so without this reset
+    // a case that ran earlier in the same binary keeps serving its master key to
+    // every later case — which silently turned the wrong-key negative below into
+    // a test that could not fail. Reset before each read so the provider is
+    // built from the key this case actually configured.
+    reset_sse_dek_provider();
+
    async_with_vars(
        [
            ("__RUSTFS_SSE_SIMPLE_CMK", Some(kms_key_b64)),
@@ -585,6 +585,7 @@ mod tests {
        let decoded = decode_node_heal_status(&encoded).expect("fixed v1 fixture should decode");
        assert_eq!(decoded.info().bitrot_start_cycle, 9);
        assert_eq!(decoded.operations.queue_length, 2);
+        assert_eq!(decoded.operations.queued_by_source.mrf, 0);
    }

    #[test]
@@ -1460,6 +1460,16 @@ fn managed_sse_domain(sse_type: SSEType) -> &'static str {
    }
 }

+/// The public `x-amz-server-side-encryption` value a managed scheme reports.
+fn managed_sse_public_header(sse_type: SSEType) -> &'static str {
+    match sse_type {
+        SSEType::SseKms => ServerSideEncryption::AWS_KMS,
+        // SSE-C never reaches the managed path; reporting AES256 keeps this
+        // total without inventing a third public value.
+        SSEType::SseS3 | SSEType::SseC => ServerSideEncryption::AES256,
+    }
+}
+
 fn canonical_kms_bucket_path(bucket: &str, key: &str) -> String {
    path_join_buf(&[bucket, key])
 }
@@ -2445,20 +2455,42 @@ async fn apply_managed_decryption_material_inner(
 ) -> Result<Option<DecryptionMaterial>, ApiError> {
    #[cfg(not(feature = "rio-v2"))]
    let _ = (bucket, key);
-    if !contains_managed_encryption_metadata(metadata) || !metadata.contains_key("x-amz-server-side-encryption") {
+    if !contains_managed_encryption_metadata(metadata) {
        return Ok(None);
    }

-    // Safe: presence is guaranteed by the contains_key check above.
-    let server_side_encryption = metadata.get("x-amz-server-side-encryption").cloned().unwrap_or_default();
-    let normalized_metadata = normalize_managed_metadata(metadata, Some(recode_minio_kms_context));
-
-    let encryption_type = match server_side_encryption.as_str() {
-        ServerSideEncryption::AES256 => SSEType::SseS3,
-        ServerSideEncryption::AWS_KMS => SSEType::SseKms,
-        _ => SSEType::SseS3,
+    let encryption_type = match metadata.get("x-amz-server-side-encryption").map(String::as_str) {
+        Some(ServerSideEncryption::AWS_KMS) => SSEType::SseKms,
+        Some(_) => SSEType::SseS3,
+        // MinIO never persists the public scheme header: `crypto.S3.CreateMetadata`
+        // writes only the `X-Minio-Internal-*` family and the public header is
+        // synthesized onto the response by `DecryptObjectInfo`. Requiring it here
+        // is what made every MinIO-encrypted object unreadable (backlog#1638).
+        //
+        // Inferring from the sealed-key slot is self-consistent by construction:
+        // the slot decides which header the unseal reads AND which domain string
+        // the sealing key is derived under, so a scheme that disagrees with the
+        // slot cannot silently derive a wrong key — it finds no key at all.
+        // Inferring from the KMS key id would NOT be safe: MinIO writes
+        // `-S3-Kms-Key-Id` on SSE-S3 objects too.
+        #[cfg(feature = "rio-v2")]
+        None => match infer_minio_managed_sse_type(metadata) {
+            Some(sse_type) => sse_type,
+            // Still fail-closed, and deliberately not an error raised here: the
+            // read plan independently classifies the object as encrypted from
+            // its markers and refuses to serve it without material, so an
+            // object whose scheme cannot be established never degrades into a
+            // plaintext read.
+            None => return Ok(None),
+        },
+        // Without the rio-v2 reader there is no MinIO-format read path to serve
+        // such an object with, so it stays on the fail-closed branch.
+        #[cfg(not(feature = "rio-v2"))]
+        None => return Ok(None),
    };

+    let normalized_metadata = normalize_managed_metadata(metadata, Some(recode_minio_kms_context));
+
    // Extract KMS key ID from metadata (optional, used for provider context)
    let kms_key_id = normalized_metadata
        .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
@@ -2556,8 +2588,19 @@ async fn apply_managed_decryption_material_inner(
    } else {
        get_local_sse_dek_provider().await?
    };
+    // A MinIO sealed key alone does not mean MinIO wrote the object: RustFS's own
+    // writer fills MinIO's metadata slots too, while still storing a RustFS
+    // envelope in them, so neither the slot nor the header name distinguishes the
+    // two. The data key's own shape does. RustFS envelopes are strictly-parsed
+    // JSON; MinIO's builtin-KMS ciphertext is opaque bytes that match neither, so
+    // recognizing RustFS positively — and treating only the remainder as MinIO —
+    // keeps a RustFS envelope from ever reaching MinIO's decoder.
    #[cfg(feature = "rio-v2")]
-    let decrypted_data_key = if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
+    let decrypted_data_key = if minio_sealed_key.is_some() && !is_rustfs_managed_data_key(&encrypted_data_key) {
+        provider
+            .decrypt_minio_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
+            .await
+    } else if is_legacy_rustfs_managed_metadata(&normalized_metadata) {
        provider
            .decrypt_legacy_sse_dek(&encrypted_data_key, &kms_key_id, &object_context)
            .await
@@ -2592,7 +2635,11 @@ async fn apply_managed_decryption_material_inner(

    Ok(Some(DecryptionMaterial {
        sse_type: encryption_type,
-        server_side_encryption: ServerSideEncryption::from(server_side_encryption),
+        // Synthesized from the resolved scheme rather than read back from
+        // metadata: a MinIO-written object has no stored scheme header, which is
+        // exactly why the gate above had to infer it. MinIO synthesizes the same
+        // header onto its own responses.
+        server_side_encryption: ServerSideEncryption::from(managed_sse_public_header(encryption_type).to_string()),
        kms_key_id: Some(SSEKMSKeyId::from(kms_key_id)),
        algorithm,
        customer_key_md5: None,
@@ -2659,6 +2706,29 @@ pub trait SseDekProvider: Send + Sync {
    ) -> Result<[u8; 32], ApiError> {
        self.decrypt_sse_dek(encrypted_dek, kms_key_id, context).await
    }
+
+    /// Unwrap a data key that MinIO's builtin KMS sealed.
+    ///
+    /// A separate entry point rather than a shape sniff inside
+    /// [`Self::decrypt_sse_dek`]: the caller already knows the object carries a
+    /// MinIO sealed key, and MinIO's raw ciphertext is unstructured bytes that
+    /// no parser can reliably tell apart from anything else. Routing on the
+    /// caller's knowledge keeps a RustFS envelope from ever reaching MinIO's
+    /// decoder, and vice versa.
+    ///
+    /// Defaults to refusing: only a provider holding the MinIO master secret
+    /// can serve these, and a provider that cannot must fail rather than fall
+    /// back to a decoder that would misread the bytes.
+    async fn decrypt_minio_sse_dek(
+        &self,
+        _encrypted_dek: &[u8],
+        _kms_key_id: &str,
+        _context: &ObjectEncryptionContext,
+    ) -> Result<[u8; 32], ApiError> {
+        Err(ApiError::from(StorageError::other(
+            "This KMS provider cannot unwrap a data key sealed by MinIO's builtin KMS",
+        )))
+    }
 }

 // ============================================================================
@@ -2797,6 +2867,163 @@ pub(crate) struct LocalSseDekProvider {

 const LOCAL_SSE_DEK_FORMAT_VERSION: u8 = 1;

+#[cfg(feature = "rio-v2")]
+/// Returns true when a managed-SSE data key is one RustFS itself wrote.
+///
+/// Both RustFS envelope shapes are strict JSON — the KMS envelope
+/// ([`rustfs_kms::is_data_key_envelope`]) and the local provider's
+/// [`LocalSseDekEnvelope`], whose `deny_unknown_fields` keeps it from accepting
+/// anything else. Recognition is deliberately positive: an unrecognized payload
+/// is left to MinIO's decoder rather than guessed at, and neither decoder is
+/// ever handed the other's format.
+fn is_rustfs_managed_data_key(encrypted_dek: &[u8]) -> bool {
+    if rustfs_kms::is_data_key_envelope(encrypted_dek) {
+        return true;
+    }
+    std::str::from_utf8(encrypted_dek)
+        .ok()
+        .is_some_and(|text| serde_json::from_str::<LocalSseDekEnvelope<'_>>(text).is_ok())
+}
+
+#[cfg(feature = "rio-v2")]
+/// Associated data MinIO binds when sealing a data key.
+///
+/// MinIO passes the object's encryption context as the AEAD's associated data,
+/// serialized as canonical JSON with sorted keys — the same canonicalization
+/// [`rustfs_kms::context_aad`] performs, which is why the context RustFS
+/// already rebuilds for the read can be reused verbatim. For SSE-S3 that
+/// context is `{bucket: "bucket/object"}`; for SSE-KMS it is whatever the
+/// request supplied, recovered from the stored MinIO context header.
+fn minio_kms_associated_data(context: &ObjectEncryptionContext) -> Result<Vec<u8>, ApiError> {
+    let mut ctx = context.encryption_context.clone();
+    ctx.entry(context.bucket.clone())
+        .or_insert_with(|| canonical_kms_bucket_path(&context.bucket, &context.object_key));
+    rustfs_kms::context_aad(&ctx)
+        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to canonicalize MinIO KMS context: {e}"))))
+}
+
+#[cfg(feature = "rio-v2")]
+/// MinIO's builtin-KMS ciphertext in its JSON encoding.
+///
+/// Deliberately its own type rather than a relaxation of
+/// [`LocalSseDekEnvelope`]: widening that envelope's `deny_unknown_fields`
+/// to admit this shape would also admit malformed RustFS envelopes, which
+/// backlog#1567 requires to keep failing closed.
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct MinioKmsCiphertextJson {
+    aead: String,
+    #[allow(
+        dead_code,
+        reason = "present in MinIO's encoding; the key is identified by metadata instead"
+    )]
+    #[serde(default)]
+    id: String,
+    iv: String,
+    nonce: String,
+    bytes: String,
+}
+
+/// Bytes of trailing randomness every MinIO builtin-KMS ciphertext carries:
+/// a 16-byte IV followed by a 12-byte nonce, *after* the sealed bytes.
+#[cfg(feature = "rio-v2")]
+const MINIO_KMS_RANDOM_LEN: usize = 28;
+#[cfg(feature = "rio-v2")]
+const MINIO_KMS_IV_LEN: usize = 16;
+
+#[cfg(feature = "rio-v2")]
+const MINIO_KMS_AEAD_AES_GCM: &str = "AES-256-GCM-HMAC-SHA-256";
+#[cfg(feature = "rio-v2")]
+const MINIO_KMS_AEAD_CHACHA20: &str = "ChaCha20Poly1305";
+
+#[cfg(feature = "rio-v2")]
+/// Unwrap a data key sealed by MinIO's builtin (static-secret) KMS.
+///
+/// The wire format is `sealed_bytes || iv[16] || nonce[12]` — the randomness
+/// trails the ciphertext rather than leading it, and MinIO's own decoder
+/// normalizes its legacy JSON encoding into exactly that byte order before
+/// opening it (`internal/kms/secret-key.go`, `parseCiphertext`). A raw
+/// (non-JSON) ciphertext is AES-256-GCM by definition there; the JSON form
+/// names its algorithm.
+///
+/// The sealing key is derived per ciphertext rather than being the master key:
+/// `HMAC-SHA256(master, iv)` for AES-256-GCM, `HChaCha20(master, iv)` for
+/// ChaCha20-Poly1305. The encryption context is bound as associated data.
+fn decrypt_minio_kms_data_key(encrypted_dek: &[u8], master_key: &[u8; 32], aad: &[u8]) -> Result<[u8; 32], ApiError> {
+    let (body, algorithm) = match std::str::from_utf8(encrypted_dek) {
+        // MinIO only treats a payload as JSON when it both starts and ends like
+        // an object, and falls back to the raw layout when it does not parse —
+        // mirrored here so a ciphertext that merely looks like JSON is not
+        // rejected outright.
+        Ok(text)
+            if text.starts_with('{')
+                && text.ends_with('}')
+                && let Ok(json) = serde_json::from_str::<MinioKmsCiphertextJson>(text) =>
+        {
+            let decode = |what: &str, value: &str| -> Result<Vec<u8>, ApiError> {
+                BASE64_STANDARD
+                    .decode(value)
+                    .map_err(|e| ApiError::from(StorageError::other(format!("Invalid MinIO KMS {what}: {e}"))))
+            };
+            let mut body = decode("ciphertext", &json.bytes)?;
+            body.extend_from_slice(&decode("iv", &json.iv)?);
+            body.extend_from_slice(&decode("nonce", &json.nonce)?);
+            (body, json.aead)
+        }
+        _ => (encrypted_dek.to_vec(), MINIO_KMS_AEAD_AES_GCM.to_string()),
+    };
+
+    if body.len() <= MINIO_KMS_RANDOM_LEN {
+        return Err(ApiError::from(StorageError::other(
+            "MinIO KMS ciphertext is too short to carry its IV and nonce",
+        )));
+    }
+    let (sealed, random) = body.split_at(body.len() - MINIO_KMS_RANDOM_LEN);
+    let (iv, nonce) = random.split_at(MINIO_KMS_IV_LEN);
+
+    let plaintext = match algorithm.as_str() {
+        MINIO_KMS_AEAD_AES_GCM => {
+            use aes_gcm::{Aes256Gcm, KeyInit, aead::Aead};
+            let mut mac = HmacSha256::new_from_slice(master_key)
+                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS sealing key derivation failed")))?;
+            mac.update(iv);
+            let sealing_key: [u8; 32] = mac.finalize().into_bytes().into();
+            let cipher = Aes256Gcm::new_from_slice(&sealing_key)
+                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS sealing key is not a valid AES-256 key")))?;
+            let nonce = aes_gcm::Nonce::try_from(nonce)
+                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS nonce is not 12 bytes")))?;
+            cipher.decrypt(&nonce, aes_gcm::aead::Payload { msg: sealed, aad })
+        }
+        MINIO_KMS_AEAD_CHACHA20 => {
+            use chacha20poly1305::{KeyInit, XChaCha20Poly1305, aead::Aead};
+            // MinIO derives this branch's key with HChaCha20 over the 16-byte
+            // IV, which is exactly XChaCha20-Poly1305's own construction, so the
+            // extended-nonce cipher does the derivation rather than hand-rolling it.
+            let mut extended = Vec::with_capacity(MINIO_KMS_IV_LEN + nonce.len());
+            extended.extend_from_slice(iv);
+            extended.extend_from_slice(nonce);
+            let cipher = XChaCha20Poly1305::new_from_slice(master_key)
+                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS master key is not a valid ChaCha20 key")))?;
+            let nonce = chacha20poly1305::XNonce::try_from(extended.as_slice())
+                .map_err(|_| ApiError::from(StorageError::other("MinIO KMS extended nonce is not 24 bytes")))?;
+            cipher.decrypt(&nonce, chacha20poly1305::aead::Payload { msg: sealed, aad })
+        }
+        other => {
+            return Err(ApiError::from(StorageError::other(format!(
+                "Unsupported MinIO KMS AEAD algorithm: {other}"
+            ))));
+        }
+    }
+    // An AEAD failure here is authentication, not a decode slip: a wrong master
+    // key, a tampered ciphertext, and an encryption context that does not match
+    // what sealed it all land here and must all fail closed.
+    .map_err(|_| ApiError::from(StorageError::other("MinIO KMS data key failed authentication")))?;
+
+    plaintext.try_into().map_err(|value: Vec<u8>| {
+        ApiError::from(StorageError::other(format!("MinIO KMS data key must be 32 bytes, got {}", value.len())))
+    })
+}
+
 #[derive(Debug, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 struct LocalSseDekEnvelope<'a> {
@@ -3013,6 +3240,17 @@ impl SseDekProvider for LocalSseDekProvider {
        let dek = Self::decrypt_dek(encrypted_dek_str, self.master_key)?;
        Ok(dek)
    }
+
+    #[cfg(feature = "rio-v2")]
+    async fn decrypt_minio_sse_dek(
+        &self,
+        encrypted_dek: &[u8],
+        _kms_key_id: &str,
+        context: &ObjectEncryptionContext,
+    ) -> Result<[u8; 32], ApiError> {
+        let aad = minio_kms_associated_data(context)?;
+        decrypt_minio_kms_data_key(encrypted_dek, &self.master_key, &aad)
+    }
 }

 // ============================================================================
@@ -3201,6 +3439,23 @@ fn is_legacy_rustfs_managed_metadata(metadata: &HashMap<String, String>) -> bool
        && !metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER)
 }

+#[cfg(feature = "rio-v2")]
+#[cfg(feature = "rio-v2")]
+/// Infer the managed SSE scheme from the MinIO sealed-key slot that is present.
+///
+/// Returns `None` when no managed MinIO slot is present, which keeps callers on
+/// their fail-closed path. SSE-C is not a managed scheme and is handled by the
+/// SSE-C read path, so its slot is not considered here.
+fn infer_minio_managed_sse_type(metadata: &HashMap<String, String>) -> Option<SSEType> {
+    if metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER) {
+        Some(SSEType::SseS3)
+    } else if metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER) {
+        Some(SSEType::SseKms)
+    } else {
+        None
+    }
+}
+
 #[cfg(feature = "rio-v2")]
 fn parse_minio_managed_sealed_key(
    metadata: &HashMap<String, String>,
@@ -117,7 +117,7 @@ pub(crate) mod access_consumer {

 pub(crate) mod concurrency_consumer {
    pub(crate) use super::super::concurrency::{
-        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard,
+        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectAdmission, PutObjectGuard,
        get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
    };
 }
Author	SHA1	Message	Date
houseme	dbef072bfe	Merge branch 'main' into overtrue/kms-1638-d2-minio-sse-read	2026-08-18 12:46:12 +08:00
Zhengchao An	deb0edb7cc	chore: adjudicate 26 bare dead_code allows across five crates (#6187 ) Remove every bare `#[allow(dead_code)]` in io-core, object-capacity, targets, rio, and scanner. Each allow was stripped first and clippy was then asked which ones the compiler actually missed, so the verdicts rest on the diagnostic rather than on inspection. 23 were inert: they sat on `pub fn`s inside `pub mod`s, where `dead_code` does not apply, or on scanner integration-test helpers that the tests in the same file do call. The remaining 3 are in rio's private `compress_index` module and the code behind them is deleted rather than annotated. `remove_index_headers` is dead and also wrong — after skipping the 4-byte chunk header it matches against `S2_INDEX_TRAILER` where `S2_INDEX_HEADER` sits, so it returns `None` for every well-formed index; rio-v2 carries the correct equivalent that is actually in use. `restore_index_headers` is its unreachable counterpart, likewise duplicated live in rio-v2. `Index::reset` is a private method with no caller. Refs backlog#1823	2026-08-18 12:45:42 +08:00
houseme	a08de9229b	feat(heal): wire MRF intents with durable repair journal (HS-01) (#6189 ) * feat(common): add MRF intent channel and Mrf request source (HS-01) Introduce the producer-facing half of the mission repair feed: a global bounded (8192) channel carrying lightweight MrfIntent values from IO error paths, plus the RUSTFS_HEAL_MRF_ENABLE delivery kill-switch and config constants for queue/journal sizing. Delivery is strictly non-blocking (try_send, drop-on-full) so it can sit on decode-failure and partial-write paths without adding latency. HealRequestSource grows a 'mrf' variant so admission accounting can attribute replayed intents. Part of backlog#1865 (option a: wire HealEvent-style intents with a durable retry ledger). Co-Authored-By: heihutu <heihutu@gmail.com> * feat(heal): add MRF queue, durable journal, and intent consumer (HS-01) Consumer half of the mission repair feed: a bounded pending queue (100k intents / 8 MiB dual ceiling, drop-newest on overflow), a durable journal at buckets/.heal/mrf/journal.bin holding the unaccepted pending snapshot, and a consumer task that batches intents off the global channel, translates them into prioritized heal requests (decode failure -> Urgent ECDecode, metadata corruption -> High Metadata, partial write -> Normal object heal), and retries full admissions with a 5s backoff and a 3-attempt ceiling. Durability: every journal record carries its own CRC32 and a format/version header, so a torn tail truncates cleanly at replay; the journal is deleted after a successful replay and when the pending set drains (mirroring MinIO's post-replay list.bin unlink). Losing the last 500 ms flush window is acceptable: replayed duplicates merge via the manager dedup key and read-repair remains the safety net. Metrics: rustfs_heal_mrf_queue_depth/_queue_bytes, _dropped_total {reason}, _replayed_total, _journal_bytes, _journal_fsync_total. The consumer is wired at heal runtime bootstrap right after manager start, honoring RUSTFS_HEAL_MRF_ENABLE (default on, rollback = off). Tests: unit tests for the dual ceiling, record roundtrip, torn-tail truncation, and the priority mapping; integration tests against a real 4-disk ECStore proving channel intents reach the manager queue as Urgent/mrf-attributed requests and journal replay arms intents, drops torn tails, and removes the file. Part of backlog#1865 (option a). Co-Authored-By: heihutu <heihutu@gmail.com> * feat(ecstore,scanner): deliver MRF intents from error paths (HS-01) Wire the three production delivery points, each a single non-blocking try_send next to the existing in-memory heal paths, which stay as the fast path: - read.rs decode-error branch: DecodeFailure intent beside the existing read-repair submit, so an Urgent ECDecode request survives restarts even when the Low-priority read-repair request was dropped or lost. - add_partial: PartialWrite intent, giving partial-write recovery a durable Normal-priority object heal across restarts. - scanner_folder metadata-corruption classification: MetadataCorruption intent beside the existing High-priority scanner heal request. All three are on error paths only: zero cost on healthy IO. Part of backlog#1865 (option a). Co-Authored-By: heihutu <heihutu@gmail.com> * fix: include mrf heal source counts Co-Authored-By: heihutu <heihutu@gmail.com> * fix: keep node heal status wire compatibility Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>	2026-08-18 12:43:27 +08:00
Zhengchao An	abffa5cf1b	chore(storage): drop dead io-schedule metrics and helpers (#6199 )	2026-08-18 04:35:36 +00:00
Zhengchao An	b825c54850	refactor(admin): route kms management auth through shared gate (#6194 )	2026-08-18 12:21:49 +08:00
houseme	de9145e87a	feat(storage): add default-off PUT admission gate (#6197 ) Add an experimental fixed-count foreground PutObject admission gate for #1882 Phase 0 validation. The gate is default-off, returns SlowDown before body ingest when saturated, and keeps the admission permit with the spawned store commit owner until store PUT returns. Co-authored-by: heihutu <heihutu@gmail.com>	2026-08-18 04:15:19 +00:00
houseme	84bd76a3ce	chore(deps): refresh cargo dependencies (#6198 ) Update workspace Cargo dependency requirements and lockfile after cargo update/upgrade, including rumqttc-next 0.34.0 and MQTT API compatibility adjustments. Verification: - cargo update --verbose - cargo upgrade --verbose - cargo update -p rumqttc-next --precise 0.34.0 --verbose - cargo tree --invert rumqttc-next --locked - cargo metadata --locked --no-deps --format-version 1 - cargo fmt --all --check - cargo check -p rustfs-targets --all-targets --locked - cargo test -p rustfs-targets mqtt --locked - make pre-pr Co-authored-by: heihutu <heihutu@gmail.com>	2026-08-18 04:11:53 +00:00
overtrue	420bfa859b	fix(sse): read objects that MinIO encrypted RustFS could not read a single MinIO-encrypted object. Two independent blockers, and backlog#1638 could only argue them statically because the fixtures the interop tests consume are generated, not checked in — so those tests had never once run. With the fixture lab working, both are now measured, fixed and covered. The detection gate required `x-amz-server-side-encryption` to be present. MinIO never persists it: `crypto.S3.CreateMetadata` writes only the `X-Minio-Internal-*` family and the public header is synthesized onto the response by `DecryptObjectInfo`. Every MinIO object therefore fell out of the managed path and failed with "encrypted object metadata is incomplete". The scheme is now inferred from which sealed-key slot is present, which is self-consistent by construction: the slot decides both which header the unseal reads and which domain string the sealing key is derived under, so an inference that disagreed with the slot could not silently derive a wrong key. Inferring from the KMS key id would NOT be safe — MinIO writes `-S3-Kms-Key-Id` on SSE-S3 objects too, which the fixtures show and a mutation test pins. Past the gate, the data key itself could not be unwrapped. Its wire format is `sealed_bytes \|\| iv[16] \|\| nonce[12]` — the randomness trails the ciphertext rather than leading it — with a per-ciphertext sealing key of `HMAC-SHA256(master, iv)` and the encryption context bound as associated data (`internal/kms/secret-key.go`). Note this is not the `{"aead":...}` JSON that backlog#1638's analysis described: current MinIO writes the raw layout and treats JSON only as a legacy encoding, normalizing it into the same byte order. Both are decoded here, in a decoder of their own — `LocalSseDekEnvelope`'s `deny_unknown_fields` is untouched, since loosening it to admit MinIO's shape would also admit malformed RustFS envelopes that backlog#1567 requires to keep failing closed. Routing between the two decoders cannot key on metadata: RustFS's own writer fills MinIO's slots while storing a RustFS envelope in them, so neither the slot nor the header name distinguishes writers. It keys on the data key's own shape instead, recognizing the two strict RustFS JSON shapes positively and leaving only the remainder to MinIO — so neither decoder is ever handed the other's format. Three round-trip tests caught an earlier slot-based attempt doing exactly that. Fail-closed is preserved throughout: a scheme that cannot be established still returns None, and the read plan independently classifies the object as encrypted from its markers and refuses to serve it without material, so no path degrades into returning ciphertext as plaintext. The interop harness also gets a provider reset. The DEK provider is cached process-wide, so a case that ran earlier kept serving its master key to every later case — which silently made the wrong-key negative test unable to fail. It fails correctly now, and the whole suite is meaningful for the first time. Refs rustfs/backlog#1638.	2026-08-18 09:32:29 +08:00