docs(listing): describe degraded strict quorum limits

fix(listing): keep the V1 NextMarker escape and document optimal limits (#8976 )
A cursor for a key that literally ends in a legacy-shaped tag is issued wrapped in one extra `[rustfs_cache:v2,return:]`, because the listing layer still strips one legacy tag from incoming markers. ListObjects (V1) unwrapped it in `calculate_next_marker`; the client then sent the bare key back as `marker`, the listing layer stripped the key's own tail, and the next page restarted at the stem and returned the same key forever. Keep the wrapper in V1 NextMarker and cover the V1 resume path (marker sent back as StartAfter) on a real store. Document that sampled list quorum modes can also omit committed objects whose copies mostly sit outside the sample, and recommend `strict` for backup, sync, replication and lifecycle tooling.
2026-10-27 14:27:10 +00:00 · 2026-10-27 09:40:46 +08:00 · 2026-10-27 08:16:34 +08:00 · 2026-10-27 06:07:52 +08:00 · 2026-10-27 03:00:18 +08:00 · 2026-10-27 02:05:58 +08:00
42 changed files with 3484 additions and 3724 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=347a0d4e5e547d5388703062ce66a66167a845b6e57ec591ccb4c732a567c68d
-sha256-linux=4f9efa47194279185bb68249e82c788fa147abe944b6c764050e3d47f31e7926
+sha256-darwin=3814cdde8cacf4187e18188d342ed70931cc32e268e42e5bb438d565bb4d0f81
+sha256-linux=663025a80273579f95ae3f49af02c0afcbfce94df09701c360a13e607b7637db
@@ -29,7 +29,7 @@
    {
      "invariant": "list-uncommitted-version",
      "suite": "rustfs-filemeta",
-      "name": "metacache::tests::resolve_with_write_quorum_slack_keeps_partial_latest_hidden_during_merge"
+      "name": "metacache::tests::resolve_merge_keeps_the_newest_version_at_listing_quorum"
    },
    {
      "invariant": "minio-object-fixture",
@@ -365,9 +365,9 @@ checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"

 [[package]]
 name = "arrow"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4c25317ebddb8f2e6f006759336c95e5ded847a446f3524c27853bfe7d77a69b"
+checksum = "3f04e0a89cc175eaf3740282537851e6fd9d7276f97730b9d0ea2859c11a85df"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -386,9 +386,9 @@ dependencies = [

 [[package]]
 name = "arrow-arith"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ab555c0ffe9cb817041c0cee80baa9663a22225d0a035693579771f2693cadc"
+checksum = "157943c720919e8f5b6731407ce5cd4c9a5ef47a75bf10edaf24b7081b5763ef"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -400,9 +400,9 @@ dependencies = [

 [[package]]
 name = "arrow-array"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2dae1d91737421e92c6dd7c7f9d0f2d25198fd6fc539738b628f4dad10adf533"
+checksum = "72b5b1682924f11ecf6929689c16cbc69efbb6dd141b06ad98ee1cb10d858c43"
 dependencies = [
 "ahash",
 "arrow-buffer",
@@ -419,9 +419,9 @@ dependencies = [

 [[package]]
 name = "arrow-buffer"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fe9cd0064f2d7cf8ecfaf77cd70e67a980bc71aea92b63102bd37a7d3379a09"
+checksum = "b5d687afc6c927965a8a759f3b68de91933d9dfa2e1bb6e156bac84b85665e92"
 dependencies = [
 "bytes",
 "half",
@@ -431,9 +431,9 @@ dependencies = [

 [[package]]
 name = "arrow-cast"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ee510bc252781d700400979a96bd1eadebeae23901132bbabd92118da1b376c5"
+checksum = "3f23c3379feede42135ef27ca60e67f6e643cd4343ae74e4c8b16c83f55afd85"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -453,9 +453,9 @@ dependencies = [

 [[package]]
 name = "arrow-csv"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ace83d5935421b950fea3a0a3f9972c0837c297ef98ade57127a5661b681112c"
+checksum = "85b1984628b091b8319b3c305fce21d04f9e9b2578bb733b0d0108462ae1d698"
 dependencies = [
 "arrow-array",
 "arrow-cast",
@@ -468,9 +468,9 @@ dependencies = [

 [[package]]
 name = "arrow-data"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "54f128c73bb7fe9f7137218f910edd5b436f390e0f040b1956f5028390787897"
+checksum = "bb0a031c7ff5e2bdb443117054fb466af1d4df5b7f0ecbe197713dc8b113b1ba"
 dependencies = [
 "arrow-buffer",
 "arrow-schema",
@@ -481,9 +481,9 @@ dependencies = [

 [[package]]
 name = "arrow-ipc"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e244f39b82fa7a823e8b5984a0bd988234f9c92c1df79378fb85090e6e4c821"
+checksum = "f5dd1afadda2eb821aaf851bb4c0adef60c30cc0f1a29bd02fb99dd07fb9f58d"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -497,9 +497,9 @@ dependencies = [

 [[package]]
 name = "arrow-json"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "727b21ebf8894dcd9504d6ab66bffdac338391e1f85cde594526a0210299ea40"
+checksum = "e2bc4d56a0dad84cc51c5812ccff48b642047a1f3be80349cd84318983e79edb"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -522,9 +522,9 @@ dependencies = [

 [[package]]
 name = "arrow-ord"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70d376b9b883239ae0f9963cdeeb59126ccaac08d60adf208a6ad3e2dd4048c3"
+checksum = "561b8209faa2cd489764b7fd07d7cb9e8d274d78048c38de59f77c41140cb834"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -535,9 +535,9 @@ dependencies = [

 [[package]]
 name = "arrow-row"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "48096811f92b5c3a0c59d33cfecd7d7d2ed037d81e54a27a42d232e7700e6de2"
+checksum = "0d8c68b87a569f92b90c46cb0a09d86bd9609ddc9b1bf1fc5a503c8c6907c14e"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -548,9 +548,9 @@ dependencies = [

 [[package]]
 name = "arrow-schema"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "37b6c8a6b12cf3869d299ce36ed8e1cd6fe9836264b68fb6c6538473a36124a1"
+checksum = "e87acb0c6bf6237fca7a50d7b51e801c98a18a5fe6f3bb34f7d77b85ca165388"
 dependencies = [
 "serde_core",
 "serde_json",
@@ -558,9 +558,9 @@ dependencies = [

 [[package]]
 name = "arrow-select"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e523e85fa9fb3181a391bf0edfa6a1ae0dc46df8cbf7f6b96034dcb1e2a7a04a"
+checksum = "c844141958d805b47d1208e0d12aac53fcf388ce8011639fe82aa492adde0bb4"
 dependencies = [
 "ahash",
 "arrow-array",
@@ -572,9 +572,9 @@ dependencies = [

 [[package]]
 name = "arrow-string"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b931a905a0e4b8832d9f54fb369cf6e545f9dc03cb3d9611216a7bd2653df60a"
+checksum = "bc130892b1a9552da77971924b390e06cf375d1f7e4344a7b7b40ca4426fa470"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -1514,6 +1514,23 @@ dependencies = [
 "serde",
 ]

+[[package]]
+name = "bindgen"
+version = "0.73.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "787ef8ef523575546b106a58213d6e6b06198a05c2f757258c68a74273670cfa"
+dependencies = [
+ "bitflags 2.13.2",
+ "cexpr",
+ "clang-sys",
+ "proc-macro2",
+ "quote",
+ "regex",
+ "rustc-hash",
+ "shlex",
+ "syn 3.0.9",
+]
+
 [[package]]
 name = "binrw"
 version = "0.15.2"
@@ -1726,13 +1743,13 @@ dependencies = [

 [[package]]
 name = "brotli"
-version = "9.0.0"
+version = "9.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f8b851b75c23ca7873623d612fe49bd1989aeb03d08fb9432187eb253d3d4c6b"
+checksum = "6737ce8b81460bf9407909f73ff50bf13f7cb8577e1e76715f57ed0eba58ba44"
 dependencies = [
 "alloc-no-stdlib 3.0.0",
 "alloc-stdlib 0.3.0",
- "brotli-decompressor 6.0.1",
+ "brotli-decompressor 6.2.0",
 ]

 [[package]]
@@ -1747,9 +1764,9 @@ dependencies = [

 [[package]]
 name = "brotli-decompressor"
-version = "6.0.1"
+version = "6.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "941cd9bd4ddab83cb46fa5a2d428f1c857b24ac78cb876cf7beb710840934bd7"
+checksum = "34c6783f42c21c952ad7112b354f0ba2f10760e9decea5b1b60523d6e270faaf"
 dependencies = [
 "alloc-no-stdlib 3.0.0",
 "alloc-stdlib 0.3.0",
@@ -1967,6 +1984,15 @@ dependencies = [
 "shlex",
 ]

+[[package]]
+name = "cexpr"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766"
+dependencies = [
+ "nom 7.1.3",
+]
+
 [[package]]
 name = "cfg-if"
 version = "1.0.5"
@@ -2083,6 +2109,17 @@ dependencies = [
 "zeroize",
 ]

+[[package]]
+name = "clang-sys"
+version = "1.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "157a8ba7b480713b56f4c09fd13fc3e0a22a5dfab8097ba61cbc5feef950788a"
+dependencies = [
+ "glob",
+ "libc",
+ "libloading",
+]
+
 [[package]]
 name = "clap"
 version = "4.6.7"
@@ -2199,7 +2236,7 @@ version = "0.4.46"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cae36683101b2899f48bd31f8db9839437356188e3b1ff23096e73dab09ba8fe"
 dependencies = [
- "brotli 9.0.0",
+ "brotli 9.1.0",
 "bzip2",
 "compression-core",
 "flate2",
@@ -2207,7 +2244,7 @@ dependencies = [
 "lz4",
 "memchr",
 "zstd 0.14.3",
- "zstd-safe 8.2.0",
+ "zstd-safe",
 ]

 [[package]]
@@ -3632,9 +3669,9 @@ dependencies = [

 [[package]]
 name = "der"
-version = "0.8.2"
+version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a878c850e9e421b20262e9b41f9c860e4785fa07541c266b62ff9d1ef998a80a"
+checksum = "6a996a108feb8b427a6e12fcf6af4aa83409b79d72c15337d7741368131ee004"
 dependencies = [
 "const-oid 0.10.2",
 "pem-rfc7468 1.0.0",
@@ -4067,7 +4104,7 @@ version = "0.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c0681a4fc24c767085329728d8dfba959af91228aa4610cca4f8ce317ba46ae0"
 dependencies = [
- "der 0.8.2",
+ "der 0.8.3",
 "digest 0.11.3",
 "elliptic-curve",
 "rfc6979",
@@ -4132,9 +4169,9 @@ checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be"

 [[package]]
 name = "elliptic-curve"
-version = "0.14.2"
+version = "0.14.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ad4496221fcfc21015fd8c0e26fcfab00504832dc0604f52acdd7bad2a8dc81"
+checksum = "900f37e7e18d74a2347389af6cd69e3a30d5cc9e697f40004865679aa75e9c7c"
 dependencies = [
 "base16ct",
 "crypto-bigint",
@@ -5236,9 +5273,9 @@ dependencies = [

 [[package]]
 name = "hotpath"
-version = "0.29.1"
+version = "0.29.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "62f2351902fa259b758c7303505273393f6b08172ab5bdd5739a2d2bc6d554e8"
+checksum = "5b4af2b5ecf2650278fe33b0e49b68c8c484d2306475c127628316dea740cf5c"
 dependencies = [
 "arc-swap",
 "async-channel",
@@ -5271,15 +5308,15 @@ dependencies = [

 [[package]]
 name = "hotpath-drain"
-version = "0.29.1"
+version = "0.29.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "83dab2340134d9976d62f9f5172f9c6584c7f521b092f258e0e58e412e160b53"
+checksum = "a5cceb6c15fbaaeb00b51acc7be7fad533d0b56cb422b1cfc16cf9ca1624d33a"

 [[package]]
 name = "hotpath-macros"
-version = "0.29.1"
+version = "0.29.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a04bbe5701b097f224a08187a78b75bb9d9b4bbaff6a279c2a50ba73edbb7aac"
+checksum = "08c0a42eb299ebb7ec6b9161489e5bf132603a34ca12cecadb8490937aa0492a"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -6399,7 +6436,7 @@ dependencies = [
 "hashbrown 0.16.1",
 "indexmap 2.14.2",
 "metrics",
- "ordered-float 5.5.0",
+ "ordered-float 5.5.1",
 "quanta",
 "radix_trie",
 "rand 0.9.5",
@@ -7333,9 +7370,9 @@ dependencies = [

 [[package]]
 name = "ordered-float"
-version = "5.5.0"
+version = "5.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8c7c9e0d9b23589f26070720bac724174bfec1083e82f7854cdd0267518343c0"
+checksum = "7f59006ce1e204c30956638535b74d317bca99477b9356dfaec54c37f6f600ab"
 dependencies = [
 "num-traits",
 ]
@@ -7452,9 +7489,9 @@ dependencies = [

 [[package]]
 name = "pageant"
-version = "0.2.4"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f0786409b5aaa884655c9b4065681db0835db0cc185a6fb19fb17a5d38656866"
+checksum = "e0bc922d9aa8cfcf598de27b3c6122cb2eeda9d6fe71527a22ebcb7f633540ff"
 dependencies = [
 "base16ct",
 "byteorder",
@@ -7523,9 +7560,9 @@ dependencies = [

 [[package]]
 name = "parquet"
-version = "59.3.1"
+version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b3d2ef0fb136e9341570f219ecb168902e76079c6d81ad555f0a4840b1e08a3"
+checksum = "8eb260d566f2fbdabe356d26aaa5d13403aac29b5b4e44463b8f9bd7b2b12680"
 dependencies = [
 "ahash",
 "arrow-array",
@@ -7810,7 +7847,7 @@ version = "0.8.0-rc.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "986d2e952779af96ea048f160fd9194e1751b4faea78bcf3ceb456efe008088e"
 dependencies = [
- "der 0.8.2",
+ "der 0.8.3",
 "spki 0.8.1",
 ]

@@ -7853,7 +7890,7 @@ dependencies = [
 "aes 0.9.3",
 "aes-gcm",
 "cbc 0.2.1",
- "der 0.8.2",
+ "der 0.8.3",
 "pbkdf2 0.13.0",
 "rand_core 0.10.1",
 "scrypt 0.12.0",
@@ -7877,7 +7914,7 @@ version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "451913da69c775a56034ea8d9003d27ee8948e12443eae7c038ba100a4f21cb7"
 dependencies = [
- "der 0.8.2",
+ "der 0.8.3",
 "pkcs5 0.8.1",
 "rand_core 0.10.1",
 "spki 0.8.1",
@@ -9139,9 +9176,9 @@ dependencies = [

 [[package]]
 name = "russh"
-version = "0.64.1"
+version = "0.65.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba61e87b9ec9a39a59a6bbed4c0b8ff7fb07073405b24a767a1f2c8560ac8f58"
+checksum = "3bb77a4b259312b2abbc541bca1c13cb4e9b61c560a7fd894068c3b92e0c9dc3"
 dependencies = [
 "aes 0.9.3",
 "aws-lc-rs",
@@ -9156,7 +9193,7 @@ dependencies = [
 "curve25519-dalek 5.0.0",
 "data-encoding",
 "delegate",
- "der 0.8.2",
+ "der 0.8.3",
 "digest 0.11.3",
 "ecdsa",
 "ed25519-dalek 3.0.0",
@@ -10847,7 +10884,7 @@ version = "1.0.2"
 dependencies = [
 "base64-simd",
 "blake2",
- "brotli 9.0.0",
+ "brotli 9.1.0",
 "bytes",
 "convert_case 0.12.0",
 "crc-fast",
@@ -11308,7 +11345,7 @@ checksum = "d56d437c2f19203ce5f7122e507831de96f3d2d4d3be5af44a0b0a09d8a80e4d"
 dependencies = [
 "base16ct",
 "ctutils",
- "der 0.8.2",
+ "der 0.8.3",
 "hybrid-array",
 "subtle",
 "zeroize",
@@ -11474,6 +11511,15 @@ dependencies = [
 "syn 3.0.9",
 ]

+[[package]]
+name = "serde_spanned"
+version = "1.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4e7523beb55eece201a2356bee0bbca0d1ab466c14c07703b2e0ee6d42cb0c2c"
+dependencies = [
+ "serde_core",
+]
+
 [[package]]
 name = "serde_urlencoded"
 version = "0.7.1"
@@ -11753,9 +11799,9 @@ checksum = "33f4fe9184a62d842c9ef383018f3306d8ba224fd9d836f56d7288308847c256"

 [[package]]
 name = "sketches-ddsketch"
-version = "0.3.1"
+version = "0.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c6f73aeb92d671e0cc4dca167e59b2deb6387c375391bc99ee743f326994a2b"
+checksum = "45f82fa313d625dd72aaa65ec0c9f8c95407add0510c1bef3c18ea7257ec34ce"

 [[package]]
 name = "slab"
@@ -11884,7 +11930,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ef958a98b9d5da290cfc78946e9f3e61e1e62a18db0d92cac0b83cc161491a9"
 dependencies = [
 "base64ct",
- "der 0.8.2",
+ "der 0.8.3",
 ]

 [[package]]
@@ -12633,6 +12679,30 @@ dependencies = [
 "webpki-roots 0.26.11",
 ]

+[[package]]
+name = "toml"
+version = "0.9.12+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf92845e79fc2e2def6a5d828f0801e29a2f8acc037becc5ab08595c7d5e9863"
+dependencies = [
+ "indexmap 2.14.2",
+ "serde_core",
+ "serde_spanned",
+ "toml_datetime 0.7.5+spec-1.1.0",
+ "toml_parser",
+ "toml_writer",
+ "winnow 0.7.15",
+]
+
+[[package]]
+name = "toml_datetime"
+version = "0.7.5+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "92e1cfed4a3038bc5a127e35a2d360f145e1f4b971b551a2ba5fd7aedf7e1347"
+dependencies = [
+ "serde_core",
+]
+
 [[package]]
 name = "toml_datetime"
 version = "1.1.2+spec-1.1.0"
@@ -12649,9 +12719,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3641d5bbb5349a79e1020a242d251efbc546ad8048d133958323ce9c40a9c9c"
 dependencies = [
 "indexmap 2.14.2",
- "toml_datetime",
+ "toml_datetime 1.1.2+spec-1.1.0",
 "toml_parser",
- "winnow",
+ "winnow 1.0.4",
 ]

 [[package]]
@@ -12660,9 +12730,15 @@ version = "1.1.5+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "baa693a8032d7e1cada7d0041e96126df243179ff061456783ac7f12bda4744c"
 dependencies = [
- "winnow",
+ "winnow 1.0.4",
 ]

+[[package]]
+name = "toml_writer"
+version = "1.1.3+spec-1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "06bdbd8cfc056b8d2e2e85f29b56a3bdbecb527cef81eb39e3e7b98af4652770"
+
 [[package]]
 name = "tonic"
 version = "0.14.6"
@@ -13689,6 +13765,12 @@ version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"

+[[package]]
+name = "winnow"
+version = "0.7.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945"
+
 [[package]]
 name = "winnow"
 version = "1.0.4"
@@ -13955,9 +14037,9 @@ dependencies = [

 [[package]]
 name = "zip"
-version = "9.1.0"
+version = "9.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2886b00003a72c4ccee3f42cfc34908d33b2508a74d827dc2a4244706475c704"
+checksum = "ec2e3161de27091f18a831d9494e9c53378a34833af5c1dda35ad85849da18d6"
 dependencies = [
 "aes 0.9.3",
 "bzip2",
@@ -14009,10 +14091,18 @@ dependencies = [
 [[package]]
 name = "zstd"
 version = "0.13.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
 dependencies = [
- "zstd-safe 7.3.0",
+ "apache-avro",
+ "arrow-array",
+ "arrow-ipc",
+ "arrow-schema",
+ "bytes",
+ "kafka-protocol",
+ "parquet",
+ "semver",
+ "toml",
+ "zstd 0.14.3",
+ "zstd-safe",
 ]

 [[package]]
@@ -14021,16 +14111,7 @@ version = "0.14.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "db3ae9bf90af6e40197bef767eeefc46ad6397420fe873830bd85e5fbbfb3c70"
 dependencies = [
- "zstd-safe 8.2.0",
-]
-
-[[package]]
-name = "zstd-safe"
-version = "7.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "64d80649ab6db9d9f6f9c80a40becd948eda4714a0a5ac8c4d157a32231c7882"
-dependencies = [
- "zstd-sys",
+ "zstd-safe",
 ]

 [[package]]
@@ -14048,6 +14129,7 @@ version = "2.1.2+zstd.1.5.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "481ada9c543b2323cbb7bbdbb5249177fbf0188ed959ed2c15e4b22421c84ed3"
 dependencies = [
+ "bindgen",
 "cc",
 "pkg-config",
 ]
@@ -66,6 +66,7 @@ members = [
    "crates/io-metrics", # I/O metrics collection for performance analysis
    "crates/io-core", # Buffered Bytes, mmap-then-copy, and aligned pread I/O helpers
    "crates/zip", # ZIP file handling and compression
+    "crates/zstd-compat", # Fixed codec for dependencies requiring zstd 0.13
 ]
 resolver = "3"

@@ -260,7 +261,7 @@ aws-smithy-http-client = { default-features = false, version = "1.5.1" }
 aws-smithy-runtime-api = { version = "1.19.1" }
 aws-smithy-types = { version = "1.9.0" }
 base64-simd = "0.8.0"
-brotli = "9.0.0"
+brotli = "9.1.0"
 clap = { version = "4.6.7" }
 const-str = { version = "1.1.0" }
 convert_case = "0.12.0"
@@ -355,7 +356,7 @@ winapi-util = "0.1.11"
 windows = { version = "0.62.2" }
 windows-sys = "0.61.2"
 xxhash-rust = { version = "0.8.19" }
-zip = "9.1.0"
+zip = "9.1.2"
 zstd = "0.14.3"

 # Observability and Metrics
@@ -376,7 +377,7 @@ libunftp = { version = "0.23.1" }
 unftp-core = "0.1.0"
 suppaftp = { version = "12.2.1" }
 rcgen = { version = "0.14.10", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] }
-russh = { version = "0.64.1" }
+russh = { version = "0.65.0" }
 russh-sftp = "3.0.1"

 # WebDAV
@@ -385,7 +386,7 @@ dav-server = "0.11.0"
 # Performance Analysis and Memory Profiling
 rustfs-mimalloc = { version = "0.7.2" }
 # Preserve Unicode focus filters until rustfs/backlog#2302 is resolved.
-hotpath = { version = "0.29.1", default-features = false }
+hotpath = { version = "0.29.2", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.49" }

@@ -425,6 +426,7 @@ strip = "none"

 # Pin upstream fixes until compatible crates.io releases are available.
 [patch.crates-io]
+zstd = { path = "crates/zstd-compat" }
 # Preserve RustFS OIDC flows while migrating the old elliptic-curve dependency chain.
 # Upstream: openidconnect-rs#244 and oauth2-rs#352.
 openidconnect = { git = "https://github.com/ramosbugs/openidconnect-rs", rev = "2f77ee73bc4554e1cb6279adb386005429fb4e41" }
@@ -20,15 +20,14 @@
 //! offline before heal converges, a strict latest-listing quorum of 3 can only
 //! ever observe 2 copies, so ListObjectsV2 silently dropped the object even
 //! though GetObject (read quorum 2) still succeeded. Exposed by the flaky
-//! "Mixed-version rolling upgrade from rc.2" CI lane (run 33478999853); the
-//! product fix relaxes the listing's required object quorum by the number of
-//! set drives the listing could not consult (see
-//! `latest_listing_required_object_quorum` in
-//! `crates/ecstore/src/store/list_objects.rs`).
+//! "Mixed-version rolling upgrade from rc.2" CI lane (run 33478999853). The
+//! listing now follows MinIO's rule: an entry needs `ceil(asked/2)` copies
+//! among the online drives it asked, so three online drives need two copies
+//! (see `SetDisks::list_path` in `crates/ecstore/src/store/list_objects.rs`).

 #[cfg(test)]
 mod tests {
-    use crate::common::{RustFSTestClusterEnvironment, init_logging};
+    use crate::common::{ClusterTopology, RustFSTestClusterEnvironment, init_logging};
    use aws_sdk_s3::Client;
    use bytes::Bytes;
    use std::collections::HashSet;
@@ -51,13 +50,13 @@ mod tests {
        format!("degraded-object-{idx:02}")
    }

-    async fn list_all_keys(client: &Client) -> Result<HashSet<String>, Box<dyn Error + Send + Sync>> {
+    async fn list_all_keys(client: &Client, bucket: &str) -> Result<HashSet<String>, Box<dyn Error + Send + Sync>> {
        let mut keys = HashSet::new();
        let mut continuation_token: Option<String> = None;
        loop {
            let response = client
                .list_objects_v2()
-                .bucket(BUCKET)
+                .bucket(bucket)
                .set_continuation_token(continuation_token.clone())
                .send()
                .await?;
@@ -147,7 +146,7 @@ mod tests {
        let expected: HashSet<String> = (0..OBJECT_COUNT).map(object_key).collect();
        let deadline = Instant::now() + LISTING_DEADLINE;
        let listed = loop {
-            let listed = match list_all_keys(&client).await {
+            let listed = match list_all_keys(&client, BUCKET).await {
                Ok(keys) => keys,
                Err(error) if Instant::now() < deadline => {
                    info!("retrying degraded listing: {error}");
@@ -171,4 +170,97 @@ mod tests {

        Ok(())
    }
+
+    /// 4-node x 4-drive cluster (one 16-drive set), #8828 shape:
+    /// 1. PUT objects while the storage class is EC:8, so each is 8+8.
+    /// 2. Restart with the EC:4 default, as after a storage-class change.
+    /// 3. Stop two nodes (8 drives). Each object keeps eight data shards, so
+    ///    GET succeeds and ListObjectsV2 must list every key: the listing
+    ///    quorum is half of the eight drives it can ask.
+    /// 4. Stop a third node. Four online drives are below the set's read
+    ///    quorum, so LIST must report 503 instead of a partial page.
+    #[tokio::test]
+    async fn ec8_objects_stay_listable_with_half_of_an_ec4_set_offline() -> TestResult {
+        const EC8_BUCKET: &str = "degraded-listing-ec8";
+        init_logging();
+
+        let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(4, 4)).await?;
+        cluster.set_env("RUSTFS_HEAL_ENABLED", "false");
+        cluster.set_env("RUSTFS_SCANNER_ENABLED", "false");
+        cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:8");
+        cluster.start().await?;
+        cluster.create_test_bucket(EC8_BUCKET).await?;
+        let client = cluster.create_s3_client(0)?;
+        for idx in 0..OBJECT_COUNT {
+            let key = object_key(idx);
+            client
+                .put_object()
+                .bucket(EC8_BUCKET)
+                .key(&key)
+                .body(Bytes::from(format!("ec8 listing payload {idx}")).into())
+                .send()
+                .await
+                .map_err(|error| format!("EC:8 PUT for {key} failed: {error}"))?;
+        }
+
+        info!("restarting with the EC:4 default storage class");
+        cluster.stop();
+        // A later entry for the same variable wins when the node process starts.
+        cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:4");
+        cluster.start().await?;
+        let client = cluster.create_s3_client(0)?;
+
+        info!("stopping nodes 2 and 3 (eight of sixteen drives)");
+        cluster.stop_node(2)?;
+        cluster.stop_node(3)?;
+        for idx in 0..OBJECT_COUNT {
+            let key = object_key(idx);
+            let deadline = Instant::now() + GET_RETRY_DEADLINE;
+            loop {
+                match client.get_object().bucket(EC8_BUCKET).key(&key).send().await {
+                    Ok(response) => {
+                        let body = response.body.collect().await?.into_bytes();
+                        assert_eq!(body.as_ref(), format!("ec8 listing payload {idx}").as_bytes());
+                        break;
+                    }
+                    Err(error) if Instant::now() < deadline => {
+                        info!("retrying EC:8 GET for {key}: {error}");
+                        tokio::time::sleep(Duration::from_millis(500)).await;
+                    }
+                    Err(error) => return Err(format!("EC:8 object {key} failed GET with eight drives: {error}").into()),
+                }
+            }
+        }
+
+        let expected: HashSet<String> = (0..OBJECT_COUNT).map(object_key).collect();
+        let deadline = Instant::now() + LISTING_DEADLINE;
+        loop {
+            match list_all_keys(&client, EC8_BUCKET).await {
+                Ok(listed) if listed == expected => break,
+                Ok(listed) => assert!(
+                    Instant::now() < deadline,
+                    "EC:8 objects readable with eight drives stayed missing from ListObjectsV2: listed={listed:?}"
+                ),
+                Err(error) => assert!(
+                    Instant::now() < deadline,
+                    "ListObjectsV2 kept failing with eight of sixteen drives online: {error}"
+                ),
+            }
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+
+        info!("stopping node 1 (four of sixteen drives remain)");
+        cluster.stop_node(1)?;
+        let deadline = Instant::now() + LISTING_DEADLINE;
+        loop {
+            match client.list_objects_v2().bucket(EC8_BUCKET).send().await {
+                Err(error) if error.raw_response().is_some_and(|response| response.status().as_u16() == 503) => break,
+                Err(error) => assert!(Instant::now() < deadline, "below the set read quorum LIST must report 503, got {error}"),
+                Ok(response) => panic!("below the set read quorum LIST must fail, but listed {} keys", response.contents().len()),
+            }
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+
+        Ok(())
+    }
 }
@@ -43,11 +43,6 @@ const EVENT_METACACHE_LISTING: &str = "metacache_listing";
 pub type AgreedFn = Box<dyn Fn(MetaCacheEntry) -> Pin<Box<dyn Future<Output = Result<(), DiskError>> + Send>> + Send + 'static>;
 pub type PartialFn =
    Box<dyn Fn(MetaCacheEntries, &[Option<DiskError>]) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
-pub(crate) type PartialResultFn = Box<
-    dyn Fn(MetaCacheEntries, &[Option<DiskError>]) -> Pin<Box<dyn Future<Output = disk::error::Result<()>> + Send>>
-        + Send
-        + 'static,
->;
 type FinishedFn = Box<dyn Fn(&[Option<DiskError>]) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;

 #[derive(Clone, Default)]
@@ -60,6 +55,7 @@ impl FallbackClaimTracker {
        self.claimed.lock().await.insert(disk.endpoint().to_string());
    }

+    #[cfg(test)]
    pub(crate) async fn claimed_keys(&self) -> HashSet<String> {
        self.claimed.lock().await.clone()
    }
@@ -317,7 +313,17 @@ pub struct ListPathRawOutcome {
 pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> {
    let rx = rx.child_token();
    let _cancel_guard = rx.clone().drop_guard();
-    list_path_raw_inner(rx, opts, None, None).await.map(|_| ())
+    list_path_raw_inner(rx, opts, None).await.map(|_| ())
+}
+
+/// Like [`list_path_raw`], but reports how the merged stream ended.
+pub(crate) async fn list_path_raw_with_outcome(
+    rx: CancellationToken,
+    opts: ListPathRawOptions,
+) -> disk::error::Result<ListPathRawOutcome> {
+    let rx = rx.child_token();
+    let _cancel_guard = rx.clone().drop_guard();
+    list_path_raw_inner(rx, opts, None).await
 }

 pub(crate) async fn list_path_raw_with_claim_tracker(
@@ -327,25 +333,13 @@ pub(crate) async fn list_path_raw_with_claim_tracker(
 ) -> disk::error::Result<ListPathRawOutcome> {
    let rx = rx.child_token();
    let _cancel_guard = rx.clone().drop_guard();
-    list_path_raw_inner(rx, opts, Some(claim_tracker), None).await
-}
-
-pub(crate) async fn list_path_raw_with_partial_result(
-    rx: CancellationToken,
-    opts: ListPathRawOptions,
-    claim_tracker: FallbackClaimTracker,
-    partial_result: PartialResultFn,
-) -> disk::error::Result<ListPathRawOutcome> {
-    let rx = rx.child_token();
-    let _cancel_guard = rx.clone().drop_guard();
-    list_path_raw_inner(rx, opts, Some(claim_tracker), Some(partial_result)).await
+    list_path_raw_inner(rx, opts, Some(claim_tracker)).await
 }

 async fn list_path_raw_inner(
    rx: CancellationToken,
    opts: ListPathRawOptions,
    fallback_claim_tracker: Option<FallbackClaimTracker>,
-    partial_result: Option<PartialResultFn>,
 ) -> disk::error::Result<ListPathRawOutcome> {
    if opts.disks.is_empty() {
        return Err(DiskError::ErasureReadQuorum);
@@ -1020,13 +1014,7 @@ async fn list_path_raw_inner(
                }
            }

-            if let Some(partial_fn) = partial_result.as_ref() {
-                tokio::select! {
-                    biased;
-                    _ = revjob_rx.cancelled() => return Ok(ListPathRawOutcome { limit_reached }),
-                    result = partial_fn(MetaCacheEntries(top_entries), &errs) => result?,
-                }
-            } else if let Some(partial_fn) = opts.partial.as_ref() {
+            if let Some(partial_fn) = opts.partial.as_ref() {
                tokio::select! {
                    biased;
                    _ = revjob_rx.cancelled() => return Ok(ListPathRawOutcome { limit_reached }),
@@ -1308,7 +1296,6 @@ mod tests {
                ..Default::default()
            },
            None,
-            None,
        )
        .await
        .expect("listing should complete");
@@ -1333,7 +1320,6 @@ mod tests {
                ..Default::default()
            },
            None,
-            None,
        )
        .await
        .expect("listing should complete");
@@ -1827,39 +1813,6 @@ mod tests {
        );
    }

-    #[tokio::test]
-    async fn list_path_raw_propagates_partial_callback_failure() {
-        let first = MetaCacheEntry {
-            name: "bucket/object-a".to_string(),
-            metadata: vec![1],
-            ..Default::default()
-        };
-        let second = MetaCacheEntry {
-            name: "bucket/object-b".to_string(),
-            metadata: vec![2],
-            ..Default::default()
-        };
-
-        let err = list_path_raw_with_partial_result(
-            CancellationToken::new(),
-            ListPathRawOptions {
-                disks: vec![None, None],
-                min_disks: 2,
-                test_reader_behaviors: vec![
-                    TestReaderBehavior::Entries(vec![first]),
-                    TestReaderBehavior::Entries(vec![second]),
-                ],
-                ..Default::default()
-            },
-            FallbackClaimTracker::default(),
-            Box::new(move |_, _| Box::pin(async { Err(DiskError::ErasureReadQuorum) })),
-        )
-        .await
-        .expect_err("a resolution failure after reader advancement must fail the listing");
-
-        assert_eq!(err, DiskError::ErasureReadQuorum);
-    }
-
    #[tokio::test]
    async fn list_path_raw_cancels_a_blocked_partial_callback() {
        let cancel = CancellationToken::new();
@@ -8748,6 +8748,89 @@ mod tests {
        peer.stop().await;
    }

+    #[tokio::test]
+    #[serial(remote_disk_recovery_probe)]
+    async fn endpoint_reconnect_admits_ready_drive_while_another_format_probe_is_blocked() {
+        runtime_sources::ensure_test_rpc_secret();
+        let reference = crate::layout::format::FormatV3::new(1, 2);
+        let mut peers = Vec::new();
+        let mut endpoints = Vec::new();
+        for index in 0..2 {
+            let mut format = reference.clone();
+            format.erasure.this = reference.erasure.sets[0][index];
+            let peer = TestGrpcPeer::spawn(
+                Bytes::from(format.to_json().expect("format should serialize")),
+                Bytes::from_static(b"recovered-peer-data"),
+            )
+            .await
+            .expect("authenticated regression peer is required");
+            endpoints.push(Endpoint {
+                url: url::Url::parse(&format!("{}/data/rustfs{index}", peer.addr)).expect("endpoint should parse"),
+                is_local: false,
+                pool_idx: 0,
+                set_idx: 0,
+                disk_idx: i32::try_from(index).expect("fixture index fits"),
+            });
+            peers.push(peer);
+        }
+        let set = crate::set_disk::SetDisks::new(
+            "independent-reconnect-test".to_owned(),
+            Arc::new(tokio::sync::RwLock::new(vec![None; 2])),
+            2,
+            1,
+            0,
+            0,
+            endpoints,
+            reference.clone(),
+            Vec::new(),
+        )
+        .await;
+        let gate = peers[0].peer.pause_next_format_read();
+        let pending_set = Arc::clone(&set);
+        let pending = tokio::spawn(async move { pending_set.connect_disks().await });
+        time::timeout(Duration::from_secs(10), gate.entered.notified())
+            .await
+            .expect("the first endpoint must reach the format barrier");
+
+        let ready = time::timeout(Duration::from_secs(10), async {
+            loop {
+                if let Some(disk) = set.disks.read().await[1].clone() {
+                    return disk;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await;
+        if let Ok(disk) = &ready {
+            assert!(set.disks.read().await[0].is_none(), "the blocked endpoint must remain unpublished");
+            assert!(!pending.is_finished(), "the sweep still owns its blocked endpoint probe");
+            assert_eq!(
+                disk.read_all("bucket", "object")
+                    .await
+                    .expect("the ready endpoint must serve real I/O"),
+                Bytes::from_static(b"recovered-peer-data")
+            );
+            assert_eq!(
+                peers[1].peer.object_read_all_disks(),
+                vec![reference.erasure.sets[0][1].to_string()],
+                "the usable handle must carry its validated slot identity"
+            );
+        }
+        gate.release.cancel();
+        pending.await.expect("all endpoint probes must finish after release");
+        for disk in set.disks.read().await.iter().flatten() {
+            disk.close().await.expect("stop each published handle's monitoring");
+        }
+        for peer in peers {
+            peer.stop().await;
+        }
+        ready.expect("a blocked earlier endpoint must not delay a ready drive in the same set");
+        assert!(
+            set.disks.read().await.iter().all(Option::is_some),
+            "both endpoints must eventually be admitted"
+        );
+    }
+
    #[tokio::test]
    #[serial(remote_disk_recovery_probe)]
    async fn missing_read_disks_preserve_concurrently_published_handles() {
@@ -1235,7 +1235,13 @@ impl DiskHealthTracker {
    }

    pub(crate) fn metric_epoch_for_reconnect(&self) -> Self {
-        Self::new()
+        let health = Self::new();
+        // Explicit test faults stay attached to the drive across handle renewal.
+        #[cfg(test)]
+        if self.test_forced_offline.load(Ordering::Acquire) {
+            health.force_offline_for_test();
+        }
+        health
    }

    pub fn last_capacity_snapshot(&self) -> Option<(u64, u64, u64, u64)> {
@@ -2664,6 +2670,11 @@ impl DiskAPI for LocalDiskWrapper {
    }

    async fn is_online(&self) -> bool {
+        // A matching format does not make a handle usable while its I/O gate is faulty.
+        // Reconnect must renew it even if its old recovery monitor has stopped.
+        if self.health.is_faulty() {
+            return false;
+        }
        let Ok(Some(disk_id)) = self.disk.get_disk_id().await else {
            return false;
        };
@@ -4695,6 +4706,38 @@ mod tests {
        assert_eq!(snapshot.total_errors_availability, 1);
    }

+    #[tokio::test]
+    async fn local_online_status_matches_io_health_and_disk_identity() {
+        let dir = tempfile::tempdir().expect("create a drive root");
+        let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 root")).expect("local endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        let meta_dir = dir.path().join(RUSTFS_META_BUCKET);
+        tokio::fs::create_dir_all(&meta_dir).await.expect("create metadata directory");
+        let mut format = FormatV3::new(1, 1);
+        format.erasure.this = format.erasure.sets[0][0];
+        tokio::fs::write(meta_dir.join(FORMAT_CONFIG_FILE), format.to_json().expect("serialize format"))
+            .await
+            .expect("persist original identity");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("open formatted drive"));
+        let wrapper = LocalDiskWrapper::new(disk, false);
+        assert!(wrapper.is_online().await, "healthy formatted drive is online");
+        wrapper.health.force_runtime_state_for_test(RuntimeDriveHealthState::Suspect);
+        assert!(wrapper.is_online().await, "suspect drives still admit foreground I/O");
+        wrapper.health.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+        assert!(!wrapper.is_online().await, "a readable format cannot make a faulty handle usable");
+        assert!(!wrapper.health.mark_recovery_success(&endpoint, "test_probe"));
+        assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Returning);
+        assert!(
+            !wrapper.is_online().await,
+            "returning drives retain the faulty I/O gate until recovery completes"
+        );
+        wrapper.reset_health_for_store_init_retry();
+        wrapper.set_disk_id_state(Some(Uuid::new_v4())).await;
+        assert!(!wrapper.is_online().await, "healthy state must not bypass a mismatched disk identity");
+    }
+
    #[tokio::test]
    async fn fresh_capacity_keeps_disk_identity_precheck_without_requiring_id_in_snapshot() {
        let dir = tempfile::tempdir().expect("temp dir should be created");
@@ -159,8 +159,18 @@ pub fn lstat_std(path: impl AsRef<Path>) -> io::Result<Metadata> {
    std::fs::symlink_metadata(path)
 }

+pub fn create_dir_all_std(path: &Path) -> io::Result<()> {
+    match std::fs::create_dir(path) {
+        Ok(()) => Ok(()),
+        Err(e) if e.kind() == io::ErrorKind::AlreadyExists => Ok(()),
+        Err(e) if e.kind() == io::ErrorKind::NotFound => std::fs::create_dir_all(path),
+        Err(e) => Err(e),
+    }
+}
+
 pub async fn make_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
-    fs::create_dir_all(path.as_ref()).await
+    let path = path.as_ref().to_path_buf();
+    tokio::task::spawn_blocking(move || create_dir_all_std(&path)).await?
 }

 fn is_dir_error(e: &io::Error) -> bool {
@@ -210,6 +220,7 @@ pub fn remove_all_std(path: impl AsRef<Path>) -> io::Result<()> {
    }
 }

+#[cfg(test)]
 pub async fn mkdir(path: impl AsRef<Path>) -> io::Result<()> {
    fs::create_dir(path.as_ref()).await
 }
@@ -239,7 +250,7 @@ impl BucketExistenceCache {
        }
    }

-    fn check_exists(&self, path: &PathBuf) -> Option<bool> {
+    fn check_exists(&self, path: &Path) -> Option<bool> {
        let mut cache = self.cache.lock().ok()?;
        if let Some((timestamp, exists)) = cache.get(path) {
            if timestamp.elapsed() < self.ttl {
@@ -256,7 +267,7 @@ impl BucketExistenceCache {
        }
    }

-    fn invalidate(&self, path: &PathBuf) {
+    fn invalidate(&self, path: &Path) {
        if let Ok(mut cache) = self.cache.lock() {
            cache.remove(path);
        }
@@ -268,15 +279,16 @@ static BUCKET_EXISTENCE_CACHE: std::sync::LazyLock<BucketExistenceCache> =

 /// Cached access check - reduces statx syscalls
 pub async fn cached_access(path: impl AsRef<Path>) -> io::Result<()> {
-    let path_buf = path.as_ref().to_path_buf();
+    let path = path.as_ref();

-    if let Some(exists) = BUCKET_EXISTENCE_CACHE.check_exists(&path_buf) {
+    if let Some(exists) = BUCKET_EXISTENCE_CACHE.check_exists(path) {
        if exists {
            return Ok(());
        }
        return Err(io::Error::new(io::ErrorKind::NotFound, "bucket not found (cached)"));
    }

+    let path_buf = path.to_path_buf();
    let result = fs::metadata(&path_buf).await;

    match &result {
@@ -292,7 +304,7 @@ pub async fn cached_access(path: impl AsRef<Path>) -> io::Result<()> {
 }

 pub fn invalidate_bucket_cache(path: impl AsRef<Path>) {
-    BUCKET_EXISTENCE_CACHE.invalidate(&path.as_ref().to_path_buf());
+    BUCKET_EXISTENCE_CACHE.invalidate(path.as_ref());
 }

 #[cfg(test)]
@@ -115,7 +115,7 @@ use time::OffsetDateTime;
 use tokio::fs::{self, File};
 #[cfg(not(unix))]
 use tokio::io::AsyncReadExt;
-use tokio::io::{AsyncRead, AsyncSeekExt, AsyncWrite, AsyncWriteExt, ErrorKind, ReadBuf};
+use tokio::io::{AsyncRead, AsyncSeekExt, AsyncWrite, ErrorKind, ReadBuf};
 use tokio::sync::{Notify, RwLock, Semaphore};
 use tokio::time::{Instant, Sleep, interval_at, timeout};
 use tracing::{debug, error, info, warn};
@@ -301,9 +301,10 @@ fn read_all_data_std(
    let size = usize::try_from(metadata.len()).map_err(|err| ReadAllError::Disk(DiskError::other(err)))?;
    let mut bytes = Vec::new();
    bytes
-        .try_reserve_exact(size)
+        .try_reserve_exact(size.saturating_add(1))
        .map_err(|err| ReadAllError::Disk(Error::other(err)))?;
-    std::io::Read::read_to_end(&mut file, &mut bytes).map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?;
+    std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, metadata.len()), &mut bytes)
+        .map_err(|err| ReadAllError::Disk(to_file_error(err).into()))?;

    let modtime = metadata.modified().ok().map(OffsetDateTime::from);
    Ok((bytes, modtime))
@@ -8379,7 +8380,7 @@ impl LocalDisk {
            return Err(DiskError::FileNotFound);
        }

-        let meta_path = path_join(&[file_path.as_ref(), Path::new(STORAGE_FORMAT_FILE)]);
+        let meta_path = file_path.as_ref().join(STORAGE_FORMAT_FILE);

        let res = {
            if read_data {
@@ -8415,8 +8416,7 @@ impl LocalDisk {
    }

    #[hotpath::measure(impl_type = "LocalDisk")]
-    async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
-        check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
+    async fn read_metadata_with_dmtime(&self, file_path: PathBuf) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
        let io_permit = acquire_context_disk_io_permit().await.map_err(DiskError::Io)?;

        // HP-12 item 1 (sub-change A): fold the open + fstat + bounded xl.meta
@@ -8429,9 +8429,7 @@ impl LocalDisk {
        //  - is_dir           -> Error::FileNotFound (NOT to_file_error(EISDIR))
        //  - metadata failure -> to_file_error
        //  - parse failure    -> propagated verbatim from read_xl_meta_no_data_sync (`?`)
-        let (data, modtime) = self
-            .run_metadata_read(io_permit, file_path.as_ref().to_path_buf(), read_metadata_only_std)
-            .await??;
+        let (data, modtime) = self.run_metadata_read(io_permit, file_path, read_metadata_only_std).await??;

        Ok((data, modtime))
    }
@@ -9462,99 +9460,62 @@ impl LocalDisk {
            skip_parent
        };

-        match data {
-            InternalBuf::Ref(buf) => {
-                let mut f = self.open_file(file_path, O_CREATE | O_WRONLY | O_TRUNC, skip_parent).await?;
-                f.write_all(buf).await.map_err(to_file_error)?;
-                if sync != SyncMode::None {
-                    f.sync_data().await.map_err(to_file_error)?;
-                    // Persist the directory entry too, so a freshly created file
-                    // survives power loss along with its contents. Skipped for
-                    // FileOnly: the caller renames the file away immediately.
-                    if sync == SyncMode::FileAndDir
-                        && let Some(parent) = file_path.parent()
-                    {
-                        os::fsync_dir(parent).await.map_err(to_file_error)?;
-                    }
-                }
-            }
-            InternalBuf::Owned(buf) => {
-                let path = file_path.to_path_buf();
-                if let Some(parent) = path.parent()
-                    && parent != skip_parent
-                {
-                    os::make_dir_all(parent, skip_parent).await?;
-                }
-
-                let root = self.publication_root.clone();
-                let operation = move |is_cancelled: &dyn Fn() -> bool| {
-                    if is_cancelled() {
-                        return Err(std::io::Error::other("blocking namespace operation cancelled"));
-                    }
-                    #[cfg(test)]
-                    run_owned_file_write_before_open(&path);
-
-                    let (mut file, parent_handle) = if sync == SyncMode::FileAndDir {
-                        root.create_truncate_with_parent(&path)?
-                    } else {
-                        (root.create_truncate(&path)?, None)
-                    };
-                    std::io::Write::write_all(&mut file, buf.as_ref())?;
-                    if sync != SyncMode::None {
-                        file.sync_data()?;
-                        // FileOnly callers rename the file away, so the tmp
-                        // directory entry never needs to become durable.
-                        if sync == SyncMode::FileAndDir
-                            && let Some(parent) = path.parent()
-                        {
-                            os::fsync_directory_handle_std(parent, parent_handle.as_ref())?;
-                        }
-                    }
-                    Ok(())
-                };
-                let result = if sync == SyncMode::None {
-                    self.blocking_queue.run_cancellable(operation).await
-                } else {
-                    self.blocking_queue.run_fsync_cancellable(operation).await
-                };
-                result.map_err(to_file_error)?;
-            }
-        }
-
-        Ok(())
-    }
-
-    async fn open_file(&self, path: impl AsRef<Path>, mode: usize, skip_parent: impl AsRef<Path>) -> Result<File> {
-        let mut skip_parent = skip_parent.as_ref();
-        if skip_parent.as_os_str().is_empty() {
-            skip_parent = self.io_root();
-        }
-
-        if let Some(parent) = path.as_ref().parent()
-            && parent != skip_parent
-        {
-            os::make_dir_all(parent, skip_parent).await?;
-        }
-
-        let f = if mode == O_CREATE | O_WRONLY | O_TRUNC {
-            let path = path.as_ref().to_path_buf();
-            let root = self.publication_root.clone();
-            File::from_std(
-                self.blocking_queue
-                    .run_cancellable(move |is_cancelled| {
-                        if is_cancelled() {
-                            return Err(std::io::Error::other("blocking namespace operation cancelled"));
-                        }
-                        root.create_truncate(&path)
-                    })
-                    .await
-                    .map_err(to_file_error)?,
-            )
-        } else {
-            super::fs::open_file(path.as_ref(), mode).await.map_err(to_file_error)?
+        #[cfg(test)]
+        let is_owned = matches!(data, InternalBuf::Owned(_));
+        let buf = match data {
+            InternalBuf::Ref(slice) => Bytes::copy_from_slice(slice),
+            InternalBuf::Owned(bytes) => bytes,
        };

-        Ok(f)
+        let path = file_path.to_path_buf();
+        let create_parent = if let Some(parent) = path.parent()
+            && parent != skip_parent
+        {
+            os::check_path_length(parent.to_string_lossy().as_ref())?;
+            !(!skip_parent.as_os_str().is_empty() && skip_parent.starts_with(parent))
+        } else {
+            false
+        };
+
+        let root = self.publication_root.clone();
+        let operation = move |is_cancelled: &dyn Fn() -> bool| {
+            if is_cancelled() {
+                return Err(std::io::Error::other("blocking namespace operation cancelled"));
+            }
+            if create_parent && let Some(parent) = path.parent() {
+                super::fs::create_dir_all_std(parent)?;
+            }
+            #[cfg(test)]
+            if is_owned {
+                run_owned_file_write_before_open(&path);
+            }
+
+            let (mut file, parent_handle) = if sync == SyncMode::FileAndDir {
+                root.create_truncate_with_parent(&path)?
+            } else {
+                (root.create_truncate(&path)?, None)
+            };
+            std::io::Write::write_all(&mut file, buf.as_ref())?;
+            if sync != SyncMode::None {
+                file.sync_data()?;
+                // FileOnly callers rename the file away, so the tmp
+                // directory entry never needs to become durable.
+                if sync == SyncMode::FileAndDir
+                    && let Some(parent) = path.parent()
+                {
+                    os::fsync_directory_handle_std(parent, parent_handle.as_ref())?;
+                }
+            }
+            Ok(())
+        };
+        let result = if sync == SyncMode::None {
+            self.blocking_queue.run_cancellable(operation).await
+        } else {
+            self.blocking_queue.run_fsync_cancellable(operation).await
+        };
+        result.map_err(to_file_error)?;
+
+        Ok(())
    }

    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
@@ -10516,10 +10477,11 @@ async fn read_file_all(path: impl AsRef<Path>, root: &os::PublicationRoot) -> Re
    let root = root.clone();
    tokio::task::spawn_blocking(move || {
        let result = (|| {
-            let mut file = root.open_readonly(&path)?;
+            let file = root.open_readonly(&path)?;
            let metadata = file.metadata()?;
-            let mut data = Vec::new();
-            std::io::Read::read_to_end(&mut file, &mut data)?;
+            let size = usize::try_from(metadata.len()).map_err(std::io::Error::other)?;
+            let mut data = Vec::with_capacity(size.saturating_add(1));
+            std::io::Read::read_to_end(&mut std::io::Read::take(file, metadata.len()), &mut data)?;
            Ok::<_, std::io::Error>((Bytes::from(data), metadata))
        })();
        result
@@ -10575,7 +10537,21 @@ fn local_disk_object_path(root: &Path, bucket: &str, key: &str) -> Result<PathBu
    Ok(path)
 }

+#[cfg(not(windows))]
+#[inline]
+fn is_clean_relative_segment(s: &str) -> bool {
+    let b = s.as_bytes();
+    !matches!(b.first(), None | Some(b'/' | b'.')) && !b.ends_with(b"/") && !b.windows(2).any(|w| matches!(w, b"//" | b"/."))
+}
+
 fn build_local_disk_object_path(root: &Path, bucket: &str, key: &str) -> (PathBuf, PathBuf) {
+    #[cfg(not(windows))]
+    if is_clean_relative_segment(bucket) && is_clean_relative_segment(key) {
+        let bucket_path = root.join(bucket);
+        let path = bucket_path.join(key);
+        return (bucket_path, path);
+    }
+
    let cache_key = if key.is_empty() {
        bucket.to_string()
    } else {
@@ -10740,9 +10716,10 @@ fn reject_local_disk_symlink_components(root: &Path, path: &Path) -> Result<()>
 fn normalize_path_cow(path: &Path) -> std::borrow::Cow<'_, Path> {
    // Windows prefixes retain the existing PathBuf reconstruction semantics.
    #[cfg(not(windows))]
-    if !path
-        .components()
-        .any(|c| matches!(c, std::path::Component::ParentDir | std::path::Component::CurDir))
+    if !path.as_os_str().as_encoded_bytes().contains(&b'.')
+        || !path
+            .components()
+            .any(|c| matches!(c, std::path::Component::ParentDir | std::path::Component::CurDir))
    {
        return std::borrow::Cow::Borrowed(path);
    }
@@ -12462,12 +12439,12 @@ impl DiskAPI for LocalDisk {
        let src_volume_dir = self.io_get_bucket_path(src_volume)?;
        let dst_volume_dir = self.io_get_bucket_path(dst_volume)?;
        if !skip_access_checks(src_volume) {
-            access(&src_volume_dir)
+            cached_access(&src_volume_dir)
                .await
                .map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
        }
        if !skip_access_checks(dst_volume) {
-            access(&dst_volume_dir)
+            cached_access(&dst_volume_dir)
                .await
                .map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
        }
@@ -12546,7 +12523,7 @@ impl DiskAPI for LocalDisk {
        if !origvolume.is_empty() {
            let origvolume_dir = self.io_get_bucket_path(origvolume)?;
            if !skip_access_checks(origvolume) {
-                access(origvolume_dir)
+                cached_access(origvolume_dir)
                    .await
                    .map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
            }
@@ -13176,7 +13153,7 @@ impl DiskAPI for LocalDisk {
        if !org_volume.is_empty() {
            let org_volume_path = self.io_get_bucket_path(org_volume)?;
            if !skip_access_checks(org_volume) {
-                access(&org_volume_path)
+                cached_access(&org_volume_path)
                    .await
                    .map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
            }
@@ -73,6 +73,10 @@ pub fn check_path_length(path_name: &str) -> Result<()> {
        return Err(DiskError::FileAccessDenied);
    }

+    if path_name.len() <= 255 {
+        return Ok(());
+    }
+
    // Check each path segment length is > 255 on all Unix
    // platforms, look for this value as NAME_MAX in
    // /usr/include/linux/limits.h
@@ -6523,16 +6527,31 @@ pub(crate) fn read_multipart_reclaim_target_metadata(
    read_multipart_metadata(path, base, publication_root, None)
 }

-fn read_multipart_metadata_bytes(mut reader: impl std::io::Read, maximum: Option<u64>) -> io::Result<Vec<u8>> {
-    let mut bytes = Vec::new();
-    match maximum {
-        Some(maximum) => {
+fn read_multipart_metadata_bytes(
+    mut reader: impl std::io::Read,
+    size_hint: Option<u64>,
+    maximum: Option<u64>,
+) -> io::Result<Vec<u8>> {
+    if let (Some(len), Some(maximum)) = (size_hint, maximum)
+        && len > maximum
+    {
+        return Err(io::Error::new(io::ErrorKind::InvalidData, "multipart metadata exceeds its size limit"));
+    }
+    let cap = size_hint
+        .and_then(|len| usize::try_from(len).ok())
+        .map_or(0, |len| len.saturating_add(1));
+    let mut bytes = Vec::with_capacity(cap);
+    match (size_hint, maximum) {
+        (Some(len), _) => {
+            std::io::Read::read_to_end(&mut reader.take(len), &mut bytes)?;
+        }
+        (None, Some(maximum)) => {
            std::io::Read::read_to_end(&mut reader.take(maximum.saturating_add(1)), &mut bytes)?;
            if bytes.len() as u64 > maximum {
                return Err(io::Error::new(io::ErrorKind::InvalidData, "multipart metadata exceeds its size limit"));
            }
        }
-        None => {
+        (None, None) => {
            std::io::Read::read_to_end(&mut reader, &mut bytes)?;
        }
    }
@@ -6564,10 +6583,11 @@ fn read_multipart_metadata(
            OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
            Mode::empty(),
        )?);
-        if !file.metadata()?.is_file() {
+        let metadata = file.metadata()?;
+        if !metadata.is_file() {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "multipart metadata is not a regular file"));
        }
-        read_multipart_metadata_bytes(&mut file, maximum)
+        read_multipart_metadata_bytes(&mut file, Some(metadata.len()), maximum)
    }
    #[cfg(windows)]
    {
@@ -6598,7 +6618,7 @@ fn read_multipart_metadata(
        if windows_file_attribute_tag(&file)?.FileAttributes & (FILE_ATTRIBUTE_DIRECTORY | FILE_ATTRIBUTE_REPARSE_POINT) != 0 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "multipart metadata is not a regular file"));
        }
-        read_multipart_metadata_bytes(file.as_file_mut(), maximum)
+        read_multipart_metadata_bytes(file.as_file_mut(), None, maximum)
    }
    #[cfg(not(any(unix, windows)))]
    {
@@ -7234,33 +7254,16 @@ pub async fn reliable_mkdir_all(path: impl AsRef<Path>, base_dir: impl AsRef<Pat
 /// This function will not create directories under base_dir.
 #[tracing::instrument(level = "debug", skip_all)]
 pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>) -> io::Result<()> {
-    if !base_dir.as_ref().to_string_lossy().is_empty() && base_dir.as_ref().starts_with(dir_path.as_ref()) {
+    let dir_path = dir_path.as_ref();
+    let base_dir = base_dir.as_ref();
+    if !base_dir.as_os_str().is_empty() && base_dir.starts_with(dir_path) {
        return Ok(());
    }

-    if let Err(e) = super::fs::mkdir(dir_path.as_ref()).await {
-        if e.kind() == io::ErrorKind::AlreadyExists {
-            return Ok(());
-        }
-
-        if e.kind() != io::ErrorKind::NotFound {
-            return Err(e);
-        }
-
-        if let Some(parent) = dir_path.as_ref().parent() {
-            // Fall back to creating the missing parent chain only when the direct mkdir proves it is required.
-            if let Err(parent_err) = super::fs::make_dir_all(parent).await
-                && parent_err.kind() != io::ErrorKind::AlreadyExists
-            {
-                return Err(parent_err);
-            }
-        }
-
-        if let Err(retry_err) = super::fs::mkdir(dir_path.as_ref()).await
-            && retry_err.kind() != io::ErrorKind::AlreadyExists
-        {
-            return Err(retry_err);
-        }
+    if let Err(err) = super::fs::make_dir_all(dir_path).await
+        && err.kind() != io::ErrorKind::AlreadyExists
+    {
+        return Err(err);
    }

    Ok(())
@@ -5232,14 +5232,10 @@ impl SetDisks {
        let disks = self.get_disks_internal().await;
        let transaction_path = part_transaction_path(dst_object);
        let transaction_meta_path = format!("{transaction_path}/{PART_TRANSACTION_NEW_META}");
-        let rollback_path = format!("{transaction_path}/{PART_TRANSACTION_ROLLBACK}");
-        let current_meta_path = format!("{dst_object}.meta");

        let reads = disks.iter().map(|disk| {
            let disk = disk.clone();
            let transaction_meta_path = transaction_meta_path.clone();
-            let rollback_path = rollback_path.clone();
-            let current_meta_path = current_meta_path.clone();
            async move {
                let Some(disk) = disk else {
                    return PartTransactionObservation {
@@ -5249,44 +5245,20 @@ impl SetDisks {
                        err: Some(DiskError::DiskNotFound),
                    };
                };
-                let transaction_meta = match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &transaction_meta_path).await {
-                    Ok(meta) => Some(meta),
-                    Err(DiskError::FileNotFound) => None,
-                    Err(err) => {
-                        return PartTransactionObservation {
-                            transaction_meta: None,
-                            current_meta: None,
-                            rollback: false,
-                            err: Some(err),
-                        };
-                    }
-                };
-                let rollback = match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &rollback_path).await {
-                    Ok(_) => true,
-                    Err(DiskError::FileNotFound) => false,
-                    Err(err) => {
-                        return PartTransactionObservation {
-                            transaction_meta,
-                            current_meta: None,
-                            rollback: false,
-                            err: Some(err),
-                        };
-                    }
-                };
-                let current_meta = match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &current_meta_path).await {
-                    Ok(meta) => Some(meta),
-                    Err(DiskError::FileNotFound | DiskError::DiskNotFound) => None,
-                    Err(_) => None,
+                let (transaction_meta, err) = match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &transaction_meta_path).await {
+                    Ok(meta) => (Some(meta), None),
+                    Err(DiskError::FileNotFound) => (None, None),
+                    Err(err) => (None, Some(err)),
                };
                PartTransactionObservation {
                    transaction_meta,
-                    current_meta,
-                    rollback,
-                    err: None,
+                    current_meta: None,
+                    rollback: false,
+                    err,
                }
            }
        });
-        let observations = join_all(reads).await;
+        let mut observations = join_all(reads).await;
        let read_errs = observations
            .iter()
            .map(|observation| observation.err.clone())
@@ -5303,6 +5275,42 @@ impl SetDisks {
            return Ok(false);
        }

+        let rollback_path = format!("{transaction_path}/{PART_TRANSACTION_ROLLBACK}");
+        let current_meta_path = format!("{dst_object}.meta");
+        let details = disks.iter().zip(observations.iter_mut()).map(|(disk, observation)| {
+            let disk = disk.clone();
+            let rollback_path = rollback_path.clone();
+            let current_meta_path = current_meta_path.clone();
+            async move {
+                if observation.err.is_some() {
+                    return;
+                }
+                let Some(disk) = disk else {
+                    observation.err = Some(DiskError::DiskNotFound);
+                    return;
+                };
+                match disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &rollback_path).await {
+                    Ok(_) => observation.rollback = true,
+                    Err(DiskError::FileNotFound) => {}
+                    Err(err) => {
+                        observation.err = Some(err);
+                        return;
+                    }
+                }
+                if let Ok(meta) = disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &current_meta_path).await {
+                    observation.current_meta = Some(meta);
+                }
+            }
+        });
+        join_all(details).await;
+        let read_errs = observations
+            .iter()
+            .map(|observation| observation.err.clone())
+            .collect::<Vec<_>>();
+        if let Some(err) = reduce_write_quorum_errs(&read_errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+            return Err(err);
+        }
+
        let mut current_counts: HashMap<Bytes, usize> = HashMap::new();
        let mut transaction_meta_values = HashSet::new();
        for observation in observations.iter().filter(|observation| observation.err.is_none()) {
@@ -36,13 +36,12 @@ use crate::disk::{
    self, BATCH_READ_VERSION_MAX_ITEMS, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp, Disk, DiskAPI,
 };
 use crate::set_disk::{
-    DiskError, DiskStore, FileInfo, HashAlgorithm, ReadOptions, SetDisks, build_inline_bitrot_readers_from_refs,
-    can_try_inline_data_shards_direct, codec_streaming_rollout_applies, coding,
-    collect_inline_data_shard_fileinfos_from_observations, file_info_is_valid_for_metadata, get_metadata_slowtail_fault_request,
-    inline_erasure_shard_file_offset, inline_erasure_shard_size, is_get_metadata_data_read_early_stop_enabled,
-    is_get_metadata_early_stop_bounded_fanout_enabled, is_get_metadata_early_stop_enabled,
-    is_get_metadata_non_inline_data_read_early_stop_enabled, is_version_early_stop_enabled, object_fits_single_block,
-    try_read_inline_data_shards_direct,
+    DiskError, DiskStore, FileInfo, HashAlgorithm, ReadOptions, SetDisks, can_try_inline_data_shards_direct,
+    codec_streaming_rollout_applies, collect_inline_data_shard_fileinfos_from_observations, file_info_is_valid_for_metadata,
+    get_metadata_slowtail_fault_request, inline_erasure_shard_file_offset, inline_erasure_shard_size,
+    is_get_metadata_data_read_early_stop_enabled, is_get_metadata_early_stop_bounded_fanout_enabled,
+    is_get_metadata_early_stop_enabled, is_get_metadata_non_inline_data_read_early_stop_enabled, is_version_early_stop_enabled,
+    object_fits_single_block,
 };
 use futures::future::join_all;
 use metrics::counter;
@@ -125,6 +124,7 @@ fn metadata_metrics_path(bucket: &str) -> &'static str {
    }
 }

+#[cfg(test)]
 pub(super) fn metadata_distribution_key(bucket: &str, object: &str) -> String {
    [bucket, object].join("/")
 }
@@ -449,36 +449,16 @@ pub(in crate::set_disk) fn bounded_metadata_fanout_order(
    total_disks: usize,
    default_parity_count: usize,
 ) -> Vec<usize> {
-    let fallback_order = || (0..total_disks).collect::<Vec<_>>();
    if default_parity_count == 0 || default_parity_count >= total_disks {
-        return fallback_order();
+        return (0..total_disks).collect();
    }

    let data_blocks = total_disks - default_parity_count;
-    let distribution_key = metadata_distribution_key(bucket, object);
-    let distribution = FileInfo::new(&distribution_key, data_blocks, default_parity_count)
-        .erasure
-        .distribution;
-    if distribution.len() != total_disks {
-        return fallback_order();
-    }
-
+    let start = FileInfo::distribution_start(bucket, object, total_disks);
+    let offset = (start + 1) % total_disks;
    let mut order = Vec::with_capacity(total_disks);
-    for block_index in 1..=data_blocks {
-        let Some(disk_index) = distribution
-            .iter()
-            .position(|distributed_block| *distributed_block == block_index)
-        else {
-            return fallback_order();
-        };
-        order.push(disk_index);
-    }
-    order.extend(
-        distribution
-            .iter()
-            .enumerate()
-            .filter_map(|(disk_index, block_index)| (*block_index > data_blocks).then_some(disk_index)),
-    );
+    order.extend((0..data_blocks).map(|i| (total_disks + i - offset) % total_disks));
+    order.extend((0..total_disks).filter(|&disk_index| (start + disk_index + 1) % total_disks >= data_blocks));
    order
 }

@@ -597,6 +577,9 @@ impl MetadataFanoutDiagnostics {
    }

    pub(in crate::set_disk) fn record(&self, path: &'static str) {
+        if !rustfs_io_metrics::get_stage_metrics_enabled() {
+            return;
+        }
        rustfs_io_metrics::record_get_object_metadata_fanout_duration(path, self.fanout_duration.as_secs_f64());
        if let Some(latency) = self.first_response_latency() {
            rustfs_io_metrics::record_get_object_first_metadata_response_latency(path, latency.as_secs_f64());
@@ -620,6 +603,9 @@ impl MetadataFanoutDiagnostics {
    }

    pub(in crate::set_disk) fn record_quorum_candidate_latency(&self, path: &'static str, read_quorum: usize) {
+        if !rustfs_io_metrics::get_stage_metrics_enabled() {
+            return;
+        }
        if let Some(latency) = self.quorum_candidate_latency(read_quorum) {
            rustfs_io_metrics::record_get_object_metadata_quorum_result(path, "reached");
            rustfs_io_metrics::record_get_object_quorum_reached_latency(path, latency.as_secs_f64());
@@ -629,15 +615,14 @@ impl MetadataFanoutDiagnostics {

 #[cfg(test)]
 pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
-    bucket: &str,
-    object: &str,
+    _bucket: &str,
+    _object: &str,
    candidate: &FileInfo,
    parts_metadata: &[FileInfo],
    disks: &[Option<DiskStore>],
 ) -> Option<&'static str> {
    verify_inline_body_from_observations(
-        bucket,
-        object,
+        _object,
        candidate,
        parts_metadata
            .iter()
@@ -645,13 +630,11 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
            .map(|(index, metadata)| (index, Some(metadata))),
        disks,
    )
-    .await
    .err()
 }

-async fn verify_inline_body_from_observations<'a>(
-    bucket: &str,
-    object: &str,
+fn verify_inline_body_from_observations<'a>(
+    _object: &str,
    candidate: &FileInfo,
    observations: impl IntoIterator<Item = (usize, Option<&'a FileInfo>)>,
    disks: &[Option<DiskStore>],
@@ -660,21 +643,18 @@ async fn verify_inline_body_from_observations<'a>(
        return Err(reason);
    }

-    let Ok(erasure) = coding::Erasure::try_new_with_options(
-        candidate.erasure.data_blocks,
-        candidate.erasure.parity_blocks,
-        candidate.erasure.block_size,
-        candidate.uses_legacy_checksum,
-    ) else {
+    let data_shards = candidate.erasure.data_blocks;
+    if data_shards == 0
+        || candidate.erasure.block_size == 0
+        || (candidate.erasure.parity_blocks > 0
+            && candidate.uses_legacy_checksum
+            && !reed_solomon_simd::ReedSolomonEncoder::supports(data_shards, candidate.erasure.parity_blocks))
+    {
        return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
-    };
-    let data_files =
-        match collect_inline_data_shard_fileinfos_from_observations(observations, candidate, erasure.data_shards, |index| {
-            disks.get(index).is_some_and(Option::is_some)
-        }) {
-            Ok(data_files) => data_files,
-            Err(reason) => return Err(reason),
-        };
+    }
+    let data_files = collect_inline_data_shard_fileinfos_from_observations(observations, candidate, data_shards, |index| {
+        disks.get(index).is_some_and(Option::is_some)
+    })?;

    let Some(part) = candidate.parts.first() else {
        return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
@@ -693,22 +673,51 @@ async fn verify_inline_body_from_observations<'a>(
        object_size,
        object_size,
        candidate.erasure.block_size,
-        erasure.data_shards,
+        data_shards,
        candidate.uses_legacy_checksum,
    );
-    let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
-    let Ok(mut readers) =
-        build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
-    else {
+    let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, data_shards, candidate.uses_legacy_checksum);
+    if object_size == 0 || read_length == 0 || read_length > shard_size || data_files.len() < data_shards {
+        return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+    }
+    let shards_needed = object_size.div_ceil(read_length);
+    if shards_needed > data_shards {
+        return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+    }
+    let Some(encoded_capacity) = read_length.checked_mul(shards_needed) else {
        return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
    };
-
    #[cfg(test)]
-    crate::set_disk::disk_call_counters::record(object, crate::set_disk::disk_call_counters::KIND_BODY_DECODE, 0);
-    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
-        Some(body) if body.len() == object_size => Ok(body),
-        _ => Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+    {
+        for file in &data_files {
+            if file.data.is_some() {
+                crate::set_disk::disk_call_counters::record_shard_open(_object, true);
+            }
+        }
+        crate::set_disk::disk_call_counters::record(_object, crate::set_disk::disk_call_counters::KIND_BODY_DECODE, 0);
    }
+
+    let hash_size = checksum_algo.size();
+    let need = hash_size + read_length;
+    let mut body = Vec::with_capacity(encoded_capacity);
+    for file in data_files.iter().take(shards_needed) {
+        let Some(raw) = file.data.as_deref() else {
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+        };
+        let Some(block) = raw.get(..need) else {
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+        };
+        let (hash, data) = block.split_at(hash_size);
+        if hash_size > 0 && checksum_algo.hash_encode(data).as_ref() != hash {
+            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+        }
+        body.extend_from_slice(data);
+        if body.len() >= object_size {
+            body.truncate(object_size);
+            return Ok(bytes::Bytes::from(body));
+        }
+    }
+    Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY)
 }

 fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Option<&'static str> {
@@ -786,16 +795,18 @@ fn data_read_inline_missing_shards_are_pending(
    fanout_order: &[usize],
    scheduled_fanout_len: usize,
 ) -> bool {
-    let Ok(erasure) = coding::Erasure::try_new_with_options(
-        candidate.erasure.data_blocks,
-        candidate.erasure.parity_blocks,
-        candidate.erasure.block_size,
-        candidate.uses_legacy_checksum,
-    ) else {
+    let data_shards = candidate.erasure.data_blocks;
+    if data_shards == 0
+        || candidate.erasure.block_size == 0
+        || !candidate.has_valid_erasure_geometry()
+        || (candidate.erasure.parity_blocks > 0
+            && candidate.uses_legacy_checksum
+            && !reed_solomon_simd::ReedSolomonEncoder::supports(data_shards, candidate.erasure.parity_blocks))
+    {
        return false;
-    };
+    }
    let distribution = &candidate.erasure.distribution;
-    let mut data_shards_seen_or_pending = vec![false; erasure.data_shards];
+    let mut data_shards_seen_or_pending = vec![false; data_shards];
    let mut missing_pending_data_shards = 0usize;

    for slot in slots {
@@ -803,7 +814,7 @@ fn data_read_inline_missing_shards_are_pending(
        let Some(&block_index) = distribution.get(disk_index) else {
            return false;
        };
-        if block_index == 0 || block_index > erasure.data_shards {
+        if block_index == 0 || block_index > data_shards {
            continue;
        }
        if !disks.get(disk_index).is_some_and(Option::is_some) {
@@ -1170,7 +1181,8 @@ impl SetDisks {
        default_parity_count: usize,
        allow_coalescing: bool,
    ) -> disk::error::Result<MetadataReadResult> {
-        let fanout_start = Instant::now();
+        let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
+        let fanout_start = stage_metrics_enabled.then(Instant::now);
        let mut slots = MetadataReadResult::pending(disks.len());
        let mut task_ids = vec![None; disks.len()];
        let mut observations = Vec::with_capacity(disks.len());
@@ -1220,7 +1232,7 @@ impl SetDisks {
            let diagnostic_context = diagnostic_context;
            let task = join_set.spawn(async move {
                let read = async move {
-                    let response_start = Instant::now();
+                    let response_start = stage_metrics_enabled.then(Instant::now);
                    let result = if let Some(disk) = disk {
                        #[cfg(test)]
                        let _fanout_task_guard = rename_fanout_barrier::task_guard(&object);
@@ -1243,7 +1255,7 @@ impl SetDisks {
                    } else {
                        Err(DiskError::DiskNotFound)
                    };
-                    (index, result, response_start.elapsed())
+                    (index, result, response_start.map_or(Duration::ZERO, |s| s.elapsed()))
                };
                #[cfg(feature = "diagnostic-read-version-caller")]
                return crate::cluster::rpc::runtime_sources::with_read_version_diagnostic_context(
@@ -1348,7 +1360,10 @@ impl SetDisks {
                },
                Err(join_error) => {
                    let err = DiskError::Unexpected;
-                    observations.push(MetadataFanoutObservation::from_error(&err, fanout_start.elapsed()));
+                    observations.push(MetadataFanoutObservation::from_error(
+                        &err,
+                        fanout_start.map_or(Duration::ZERO, |s| s.elapsed()),
+                    ));
                    if let Some(index) = task_ids.iter().position(|id| *id == Some(join_error.id())) {
                        slots[index].result = Some(Err(err));
                        accumulator.observe(&slots[index]);
@@ -1384,14 +1399,11 @@ impl SetDisks {
                            accumulator.candidate_has_read_reserve()
                        }
                        Some(candidate) => match verify_inline_body_from_observations(
-                            bucket.as_ref(),
                            object.as_ref(),
                            candidate,
                            slots.iter().map(|slot| (slot.disk_index, slot.file_info())),
                            disks,
-                        )
-                        .await
-                        {
+                        ) {
                            Ok(body) => {
                                verified_inline_body = Some(body);
                                true
@@ -1449,7 +1461,8 @@ impl SetDisks {
                        scheduled_count.saturating_sub(cancelled_count),
                        cancelled_count,
                    );
-                    let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations);
+                    let diagnostics =
+                        MetadataFanoutDiagnostics::new(fanout_start.map_or(Duration::ZERO, |s| s.elapsed()), observations);
                    return Ok(MetadataReadResult {
                        slots,
                        diagnostics,
@@ -1532,7 +1545,7 @@ impl SetDisks {
        rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, final_miss_reason);
        rustfs_io_metrics::record_get_object_metadata_early_stop_saved_responses(metrics_path, 0);
        rustfs_io_metrics::record_get_object_metadata_fanout_lifecycle(metrics_path, scheduled_count, scheduled_count, 0);
-        let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations);
+        let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.map_or(Duration::ZERO, |s| s.elapsed()), observations);
        Ok(MetadataReadResult {
            slots,
            diagnostics,
@@ -275,6 +275,22 @@ impl SetDisks {

    pub(super) fn common_parity(parities: &[i32], default_parity_count: i32) -> i32 {
        let n = parities.len() as i32;
+        let mut valid_parities = parities.iter().copied().filter(|&p| p != -1);
+        let Some(first_parity) = valid_parities.next() else {
+            return -1;
+        };
+        let mut same_count = 1i32;
+        if valid_parities.all(|p| {
+            same_count += 1;
+            p == first_parity
+        }) {
+            let read_quorum = if default_parity_count > 0 && first_parity == 0 {
+                n / 2 + 1
+            } else {
+                n - first_parity
+            };
+            return if same_count >= read_quorum { first_parity } else { -1 };
+        }

        let mut occ_map: HashMap<i32, i32> = HashMap::new();
        for &p in parities {
@@ -333,6 +349,19 @@ impl SetDisks {
    }

    pub(super) fn common_time_and_occurrence(times: &[Option<OffsetDateTime>]) -> (Option<OffsetDateTime>, usize) {
+        let mut valid_times = times.iter().flatten().copied();
+        let Some(first_time) = valid_times.next() else {
+            return (None, 0);
+        };
+        let mut count = 1usize;
+        if valid_times.all(|t| {
+            count += 1;
+            t == first_time
+        }) && first_time.unix_timestamp_nanos() != 0
+        {
+            return (Some(first_time), count);
+        }
+
        let mut time_occurrence_map = HashMap::new();

        // Ignore the uuid sentinel and count the rest.
@@ -599,6 +628,25 @@ impl SetDisks {
            return write_quorum;
        }

+        if let Some((first, _)) = parts_metadata
+            .iter()
+            .zip(errs.iter())
+            .find(|(meta, err)| err.is_none() && file_info_is_valid_for_metadata(meta))
+        {
+            let same_count = parts_metadata
+                .iter()
+                .zip(errs.iter())
+                .filter(|(meta, err)| {
+                    err.is_none()
+                        && file_info_is_valid_for_metadata(meta)
+                        && super::core::metadata_quorum::metadata_early_stop_candidate_matches(first, meta)
+                })
+                .count();
+            if same_count >= write_quorum {
+                return write_quorum;
+            }
+        }
+
        let mut identity_counts = HashMap::with_capacity(usable_metadata);
        for (meta, err) in parts_metadata.iter().zip(errs.iter()) {
            if err.is_some() || !file_info_is_valid_for_metadata(meta) {
@@ -982,6 +1030,33 @@ impl SetDisks {
            return Err(DiskError::ErasureReadQuorum);
        }

+        let matches_time_or_etag = |meta: &FileInfo| {
+            mod_time == &meta.mod_time
+                || (mod_time.is_none() && etag.as_deref().is_some_and(|e| meta.get_etag().as_deref() == Some(e)))
+        };
+
+        if let Some(first) = metas
+            .iter()
+            .find(|meta| file_info_is_valid_for_metadata(meta) && matches_time_or_etag(meta))
+        {
+            let mut total_valid = 0usize;
+            let mut same_count = 0usize;
+            for meta in metas {
+                if !file_info_is_valid_for_metadata(meta) || !matches_time_or_etag(meta) {
+                    continue;
+                }
+                total_valid += 1;
+                if super::core::metadata_quorum::metadata_early_stop_candidate_matches(first, meta) {
+                    same_count += 1;
+                }
+            }
+            if same_count >= quorum && same_count > total_valid - same_count {
+                let mut fi = first.clone();
+                fi.is_latest = fi.successor_mod_time.is_none();
+                return Ok(fi);
+            }
+        }
+
        let mut meta_hashes = vec![None; metas.len()];

        for (i, meta) in metas.iter().enumerate() {
@@ -160,7 +160,7 @@ use rustfs_utils::http::{
 use rustfs_utils::{
    HashAlgorithm,
    crypto::hex,
-    path::{SLASH_SEPARATOR, encode_dir_object, encode_dir_object_ref, has_suffix, object_name_disk_identity, path_join_buf},
+    path::{SLASH_SEPARATOR, encode_dir_object, has_suffix, object_name_disk_identity, path_join_buf},
 };
 use sha2::Sha256;
 use std::borrow::Cow;
@@ -934,12 +934,20 @@ pub(in crate::set_disk) struct GetObjectFileInfo {
 }

 struct OwnedGetObjectFileInfo {
-    fi: FileInfo,
-    parts_metadata: Vec<FileInfo>,
-    online_disks: Vec<Option<DiskStore>>,
+    payload: OwnedGetObjectFileInfoPayload,
    verified_inline_body: Option<Bytes>,
 }

+#[allow(clippy::large_enum_variant)]
+enum OwnedGetObjectFileInfoPayload {
+    Direct {
+        fi: FileInfo,
+        parts_metadata: Vec<FileInfo>,
+        online_disks: Vec<Option<DiskStore>>,
+    },
+    Published(Arc<GetObjectMetadataCacheEntry>),
+}
+
 enum LateMetadataFanout {
    CapturedDisks(Vec<Option<DiskStore>>),
    RefreshCurrentSet,
@@ -951,9 +959,11 @@ impl GetObjectFileInfo {
    fn owned(fi: FileInfo, parts_metadata: Vec<FileInfo>, online_disks: Vec<Option<DiskStore>>) -> Self {
        Self {
            owned: Some(OwnedGetObjectFileInfo {
-                fi,
-                parts_metadata,
-                online_disks,
+                payload: OwnedGetObjectFileInfoPayload::Direct {
+                    fi,
+                    parts_metadata,
+                    online_disks,
+                },
                verified_inline_body: None,
            }),
            shared: None,
@@ -962,15 +972,21 @@ impl GetObjectFileInfo {
        }
    }

-    fn owned_with_late_metadata_fanout(
-        fi: FileInfo,
-        parts_metadata: Vec<FileInfo>,
-        online_disks: Vec<Option<DiskStore>>,
-        late_metadata_fanout_disks: Vec<Option<DiskStore>>,
-    ) -> Self {
-        let mut snapshot = Self::owned(fi, parts_metadata, online_disks);
-        snapshot.late_metadata_fanout = Some(LateMetadataFanout::CapturedDisks(late_metadata_fanout_disks));
-        snapshot
+    fn owned_published(entry: Arc<GetObjectMetadataCacheEntry>) -> Self {
+        let condition_stopped = entry.capability == GetObjectMetadataCacheCapability::ConditionOnly;
+        Self {
+            owned: Some(OwnedGetObjectFileInfo {
+                payload: OwnedGetObjectFileInfoPayload::Published(entry),
+                verified_inline_body: None,
+            }),
+            shared: None,
+            condition_stopped,
+            late_metadata_fanout: None,
+        }
+    }
+
+    fn set_late_metadata_fanout_disks(&mut self, late_metadata_fanout_disks: Vec<Option<DiskStore>>) {
+        self.late_metadata_fanout = Some(LateMetadataFanout::CapturedDisks(late_metadata_fanout_disks));
    }

    fn shared(entry: Arc<GetObjectMetadataCacheEntry>) -> Self {
@@ -989,28 +1005,31 @@ impl GetObjectFileInfo {
        snapshot
    }

-    fn fi(&self) -> &FileInfo {
+    fn as_parts(&self) -> (&FileInfo, &[FileInfo], &[Option<DiskStore>]) {
        match (&self.owned, &self.shared) {
-            (Some(snapshot), None) => &snapshot.fi,
-            (None, Some(entry)) => &entry.fi,
+            (Some(snapshot), None) => match &snapshot.payload {
+                OwnedGetObjectFileInfoPayload::Direct {
+                    fi,
+                    parts_metadata,
+                    online_disks,
+                } => (fi, parts_metadata, online_disks),
+                OwnedGetObjectFileInfoPayload::Published(entry) => (&entry.fi, &entry.parts_metadata, &entry.online_disks),
+            },
+            (None, Some(entry)) => (&entry.fi, &entry.parts_metadata, &entry.online_disks),
            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
        }
    }

+    fn fi(&self) -> &FileInfo {
+        self.as_parts().0
+    }
+
    fn parts_metadata(&self) -> &[FileInfo] {
-        match (&self.owned, &self.shared) {
-            (Some(snapshot), None) => &snapshot.parts_metadata,
-            (None, Some(entry)) => &entry.parts_metadata,
-            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
-        }
+        self.as_parts().1
    }

    fn online_disks(&self) -> &[Option<DiskStore>] {
-        match (&self.owned, &self.shared) {
-            (Some(snapshot), None) => &snapshot.online_disks,
-            (None, Some(entry)) => &entry.online_disks,
-            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
-        }
+        self.as_parts().2
    }

    fn has_late_metadata_fanout(&self) -> bool {
@@ -1024,17 +1043,28 @@ impl GetObjectFileInfo {
    }

    fn take_verified_inline_body(&mut self) -> Option<Bytes> {
-        self.owned.as_mut().and_then(|owned| owned.verified_inline_body.take())
+        match (&mut self.owned, &self.shared) {
+            (Some(owned), None) => owned.verified_inline_body.take().or_else(|| match &owned.payload {
+                OwnedGetObjectFileInfoPayload::Direct { .. } => None,
+                OwnedGetObjectFileInfoPayload::Published(entry) => entry.verified_inline_body.clone(),
+            }),
+            (None, Some(entry)) => entry.verified_inline_body.clone(),
+            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
+        }
+    }
+
+    fn into_payload(self) -> (OwnedGetObjectFileInfoPayload, Option<LateMetadataFanout>) {
+        match (self.owned, self.shared) {
+            (Some(snapshot), None) => (snapshot.payload, self.late_metadata_fanout),
+            (None, Some(entry)) => (OwnedGetObjectFileInfoPayload::Published(entry), self.late_metadata_fanout),
+            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
+        }
    }

    fn into_object_info(self, bucket: &str, object: &str, versioned: bool) -> ObjectInfo {
-        match (self.owned, self.shared) {
-            (Some(snapshot), None) => ObjectInfo::from_owned_file_info(snapshot.fi, bucket, object, versioned),
-            (None, Some(entry)) => match Arc::try_unwrap(entry) {
-                Ok(entry) => ObjectInfo::from_owned_file_info(entry.fi, bucket, object, versioned),
-                Err(entry) => ObjectInfo::from_file_info(&entry.fi, bucket, object, versioned),
-            },
-            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
+        match self.into_payload().0 {
+            OwnedGetObjectFileInfoPayload::Direct { fi, .. } => ObjectInfo::from_owned_file_info(fi, bucket, object, versioned),
+            OwnedGetObjectFileInfoPayload::Published(entry) => ObjectInfo::from_file_info(&entry.fi, bucket, object, versioned),
        }
    }

@@ -1044,24 +1074,17 @@ impl GetObjectFileInfo {
    }

    fn into_owned_with_late_metadata_fanout(self) -> OwnedGetObjectFileInfoParts {
-        let Self {
-            owned,
-            shared,
-            late_metadata_fanout,
-            ..
-        } = self;
-        match (owned, shared) {
-            (Some(snapshot), None) => (snapshot.fi, snapshot.parts_metadata, snapshot.online_disks, late_metadata_fanout),
-            (None, Some(entry)) => match Arc::try_unwrap(entry) {
-                Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks, late_metadata_fanout),
-                Err(entry) => (
-                    entry.fi.clone(),
-                    entry.parts_metadata.clone(),
-                    entry.online_disks.clone(),
-                    late_metadata_fanout,
-                ),
-            },
-            _ => unreachable!("GET metadata snapshot representation must be exclusive"),
+        let (payload, late_metadata_fanout) = self.into_payload();
+        match payload {
+            OwnedGetObjectFileInfoPayload::Direct {
+                fi,
+                parts_metadata,
+                online_disks,
+            } => (fi, parts_metadata, online_disks, late_metadata_fanout),
+            OwnedGetObjectFileInfoPayload::Published(entry) => {
+                let entry = Arc::unwrap_or_clone(entry);
+                (entry.fi, entry.parts_metadata, entry.online_disks, late_metadata_fanout)
+            }
        }
    }

@@ -1446,6 +1469,7 @@ mod prepared_get_object_metadata_tests {
            fi,
            online_disks: vec![None],
            read_quorum: 0,
+            verified_inline_body: None,
        });
        let snapshot = GetObjectFileInfo::shared(Arc::clone(&cached));

@@ -2762,7 +2786,9 @@ impl SetDisks {
        object: &str,
        opts: &ObjectOptions,
    ) -> Result<PreparedGetObjectMetadata> {
-        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, false).await?;
+        let snapshot = self
+            .get_object_fileinfo_for_object_read(bucket, object, opts, true, false)
+            .await?;
        let object_info = build_get_object_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
        Ok(PreparedGetObjectMetadata {
            snapshot,
@@ -4901,7 +4927,7 @@ enum GetObjectMetadataCacheCapability {
    ConditionOnly,
 }

-#[derive(Debug)]
+#[derive(Clone, Debug)]
 struct GetObjectMetadataCacheEntry {
    #[allow(dead_code)] // Kept for debugging; moka handles TTL internally
    created_at: Instant,
@@ -4910,6 +4936,7 @@ struct GetObjectMetadataCacheEntry {
    online_disks: Vec<Option<DiskStore>>,
    read_quorum: usize,
    capability: GetObjectMetadataCacheCapability,
+    verified_inline_body: Option<Bytes>,
 }

 #[derive(Clone, Debug)]
@@ -5220,40 +5247,6 @@ impl SetDisks {
        .with_attempt(attempt))
    }

-    pub(crate) async fn read_listing_metadata_after_namespace_barrier(
-        &self,
-        disks: &[Option<DiskStore>],
-        bucket: &str,
-        object: &str,
-    ) -> Result<(MetaCacheEntries, usize)> {
-        // This is only called after the lock-free LIST fast path observes an
-        // unresolved metadata generation. Waiting on the ordinary object read
-        // lock establishes a publication boundary with concurrent overwrites.
-        let stored_object = encode_dir_object_ref(object);
-        let _guard = self
-            .acquire_read_lock_diag("list_object_reconcile", bucket, stored_object.as_ref())
-            .await?;
-        let (raw_entries, errors) = Self::read_all_raw_file_info(disks, bucket, stored_object.as_ref(), false).await;
-        let confirmed_absent = errors
-            .iter()
-            .flatten()
-            .filter(|err| DiskError::is_err_object_not_found(err) || DiskError::is_err_version_not_found(err))
-            .count();
-        let entries = raw_entries
-            .into_iter()
-            .map(|raw| {
-                raw.filter(|raw| !raw.buf.is_empty()).map(|raw| MetaCacheEntry {
-                    name: object.to_owned(),
-                    metadata: raw.buf,
-                    cached: None,
-                    reusable: false,
-                })
-            })
-            .collect();
-
-        Ok((MetaCacheEntries(entries), confirmed_absent))
-    }
-
    async fn acquire_write_lock_diag(&self, op: &'static str, bucket: &str, object: &str) -> Result<ObjectLockDiagGuard> {
        crate::hp_guard!("SetDisks::acquire_write_lock");
        let diag_enabled = is_object_lock_diag_enabled();
@@ -16450,54 +16443,6 @@ mod tests {
        assert_eq!(err, StorageError::PreconditionFailed);
    }

-    #[tokio::test]
-    async fn listing_metadata_reconcile_encodes_directory_object_and_waits_for_writer() {
-        let set_disks = make_local_bucket_test_set_disks().await;
-        let bucket = "bucket-list-reconcile-lock";
-        let object = "directory/";
-        let stored_object = encode_dir_object(object);
-        set_disks
-            .make_bucket(bucket, &MakeBucketOptions::default())
-            .await
-            .expect("bucket should be created");
-        let mut reader = PutObjReader::from_vec(b"directory marker".to_vec());
-        set_disks
-            .put_object(bucket, &stored_object, &mut reader, &ObjectOptions::default())
-            .await
-            .expect("directory object should be written under its encoded key");
-        let disks = set_disks.disks.read().await.clone();
-        let namespace_lock = set_disks
-            .new_ns_lock(bucket, &stored_object)
-            .await
-            .expect("namespace lock should be created");
-        let writer_guard = namespace_lock
-            .get_write_lock(std::time::Duration::from_secs(30))
-            .await
-            .expect("writer lock should be acquired");
-        let read = set_disks.read_listing_metadata_after_namespace_barrier(&disks, bucket, object);
-        tokio::pin!(read);
-        assert!(
-            tokio::time::timeout(std::time::Duration::from_millis(50), &mut read)
-                .await
-                .is_err(),
-            "LIST reconciliation must wait for the publishing writer"
-        );
-        drop(writer_guard);
-
-        let (entries, confirmed_absent) = tokio::time::timeout(std::time::Duration::from_secs(30), read)
-            .await
-            .expect("LIST reconciliation should resume after writer release")
-            .expect("LIST reconciliation should read after the namespace barrier");
-        assert_eq!(entries.0.len(), disks.len());
-        assert!(
-            entries.0.iter().all(|entry| entry
-                .as_ref()
-                .is_some_and(|entry| entry.name == object && !entry.metadata.is_empty())),
-            "LIST reconciliation must read encoded metadata and retain the logical key"
-        );
-        assert_eq!(confirmed_absent, 0);
-    }
-
    #[tokio::test]
    async fn repeated_body_write_keeps_etag_but_changes_data_dir_generation() {
        let set_disks = make_local_bucket_test_set_disks().await;
@@ -26,7 +26,9 @@ use super::super::{
 };
 use crate::disk::health_state::DriveMembershipSnapshot;
 use crate::disk::{DiskAPI, new_disk};
+use crate::layout::disks_layout::MAX_ERASURE_SET_DRIVE_COUNT;
 use crate::runtime::sources as runtime_sources;
+use futures::{StreamExt, stream};
 use rand::prelude::SliceRandom;
 use rustfs_utils::path::object_name_disk_identity;
 #[cfg(test)]
@@ -326,35 +328,31 @@ impl SetDisks {
    }

    async fn connect_disks_with_active_monitoring(&self, active_monitoring: bool) {
-        let rl = self.disks.read().await;
+        let disks = self.get_disks_internal().await;
+        // A slow failed endpoint must not hold back other returning drives.
+        // Keep the fanout within one set and release the inventory lock before I/O.
+        stream::iter(disks.into_iter().enumerate())
+            .for_each_concurrent(MAX_ERASURE_SET_DRIVE_COUNT, |(i, opdisk)| async move {
+                if let Some(disk) = opdisk {
+                    if !active_monitoring
+                        && disk.get_disk_location().set_idx.is_some()
+                        && let disk::Disk::Local(local) = disk.as_ref()
+                        && local.can_skip_background_probe().await
+                    {
+                        return;
+                    }
+                    if disk.is_online().await && disk.get_disk_location().set_idx.is_some() {
+                        return;
+                    }

-        let disks = rl.clone();
-
-        // Explicitly release the lock
-        drop(rl);
-
-        for (i, opdisk) in disks.iter().enumerate() {
-            if let Some(disk) = opdisk {
-                if !active_monitoring
-                    && disk.get_disk_location().set_idx.is_some()
-                    && let disk::Disk::Local(local) = disk.as_ref()
-                    && local.can_skip_background_probe().await
-                {
-                    continue;
-                }
-                if disk.is_online().await && disk.get_disk_location().set_idx.is_some() {
-                    info!("Disk {:?} is online", disk.to_string());
-                    continue;
+                    let _ = disk.close().await;
                }

-                let _ = disk.close().await;
-            }
-
-            if let Some(endpoint) = self.set_endpoints.get(i) {
-                info!("will renew disk, opdisk: {:?}", opdisk);
-                self.renew_disk(endpoint).await;
-            }
-        }
+                if let Some(endpoint) = self.set_endpoints.get(i) {
+                    self.renew_disk(endpoint).await;
+                }
+            })
+            .await;
    }

    pub async fn renew_disk(&self, ep: &Endpoint) {
@@ -727,10 +725,6 @@ mod tests {
        let format = FormatV3::new(1, 1);
        let (_dir, endpoint, disk) = make_formatted_local_disk(0, &format).await;
        assert!(disk.is_online().await, "initialize the expected disk identity");
-        let disk::Disk::Local(wrapper) = disk.as_ref() else {
-            panic!("fixture must use a local disk");
-        };
-        let local = wrapper.get_disk();
        let set = SetDisks::new(
            "quiet-monitor".to_owned(),
            Arc::new(RwLock::new(vec![Some(disk.clone())])),
@@ -749,13 +743,136 @@ mod tests {
            (false, RuntimeDriveHealthState::Offline),
            (false, RuntimeDriveHealthState::Returning),
        ] {
+            let disk = set.get_disks_internal().await[0].clone().expect("current drive");
+            let disk::Disk::Local(wrapper) = disk.as_ref() else {
+                panic!("fixture must use a local disk");
+            };
+            let local = wrapper.get_disk();
            disk.force_runtime_state_for_test(state);
            local.format_info.write().await.last_check = None;
            set.connect_disks_with_active_monitoring(active).await;
+            let current = set.get_disks_internal().await[0].clone().expect("probed or renewed drive");
+            let disk::Disk::Local(wrapper) = current.as_ref() else {
+                panic!("fixture must use a local disk");
+            };
            assert!(
-                local.format_info.read().await.last_check.is_some(),
+                wrapper.get_disk().format_info.read().await.last_check.is_some(),
                "probe required for active={active}, state={state:?}"
            );
+            current
+                .disk_info(&DiskInfoOptions::default())
+                .await
+                .expect("the probed or renewed drive must admit real I/O");
+        }
+    }
+
+    #[tokio::test]
+    async fn partial_reconnect_restores_ec4_io_without_all_disks_returning() {
+        use crate::object_api::{ObjectOptions, PutObjReader};
+        use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
+        use crate::storage_api_contracts::object::ObjectIO;
+        use disk::health_state::RuntimeDriveHealthState;
+        use http::HeaderMap;
+        use tokio::io::AsyncReadExt;
+
+        for active_monitoring in [true, false] {
+            let format = FormatV3::new(1, 16);
+            let mut dirs = Vec::new();
+            let mut endpoints = Vec::new();
+            let mut disks = Vec::new();
+            for index in 0..16 {
+                let (dir, endpoint, disk) = make_formatted_local_disk(index, &format).await;
+                assert!(disk.is_online().await, "initialize the original drive identity");
+                dirs.push(dir);
+                endpoints.push(endpoint);
+                disks.push(Some(disk));
+            }
+            let set = SetDisks::new(
+                "partial-reconnect".to_owned(),
+                Arc::new(RwLock::new(disks.clone())),
+                16,
+                4,
+                0,
+                0,
+                endpoints,
+                format,
+                Vec::new(),
+            )
+            .await;
+            let bucket = "partial-reconnect";
+            let opts = ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            };
+            set.make_bucket(bucket, &MakeBucketOptions::default())
+                .await
+                .expect("create bucket before the outage");
+            let payload = vec![b'r'; 1_300_000];
+            set.put_object(bucket, "before", &mut PutObjReader::from_vec(payload.clone()), &opts)
+                .await
+                .expect("write the original object");
+
+            let mut saved_formats = Vec::new();
+            for (dir, disk) in dirs.iter().zip(&disks).take(8) {
+                let disk = disk.as_ref().expect("original disk");
+                disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+                let path = dir.path().join(disk::RUSTFS_META_BUCKET).join(disk::FORMAT_CONFIG_FILE);
+                saved_formats.push(tokio::fs::read(&path).await.expect("save original format"));
+                // A corrupt format makes reconnect fail without admitting a blank replacement.
+                tokio::fs::write(&path, b"invalid-format")
+                    .await
+                    .expect("make the original drive unavailable");
+                let disk::Disk::Local(local) = disk.as_ref() else {
+                    panic!("fixture must use local drives");
+                };
+                local.get_disk().format_info.write().await.last_check = None;
+            }
+            set.connect_disks_with_active_monitoring(active_monitoring).await;
+            let unavailable = set.get_disks_internal().await;
+            for (original, retained) in disks.iter().zip(&unavailable).take(8) {
+                assert!(
+                    Arc::ptr_eq(original.as_ref().expect("original"), retained.as_ref().expect("retained")),
+                    "failed reconnect must leave the old handle to exercise subsequent recovery"
+                );
+            }
+            assert!(
+                set.put_object(bucket, "during", &mut PutObjReader::from_vec(payload.clone()), &opts)
+                    .await
+                    .is_err(),
+                "eight available drives must not satisfy a 16-drive EC:4 write"
+            );
+
+            for (dir, saved) in dirs.iter().zip(&saved_formats).take(4) {
+                let path = dir.path().join(disk::RUSTFS_META_BUCKET).join(disk::FORMAT_CONFIG_FILE);
+                tokio::fs::write(path, saved)
+                    .await
+                    .expect("restore four original drive identities");
+            }
+            set.connect_disks_with_active_monitoring(active_monitoring).await;
+            assert_eq!(
+                set.drive_membership_snapshot().await.online.len(),
+                12,
+                "the four restored drives must be usable while four others remain unavailable"
+            );
+            set.put_object(bucket, "after", &mut PutObjReader::from_vec(payload.clone()), &opts)
+                .await
+                .expect("write at the restored EC:4 quorum");
+            for object in ["before", "after"] {
+                let mut reader = set
+                    .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
+                    .await
+                    .expect("read at the restored EC:4 quorum");
+                let mut actual = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut actual)
+                    .await
+                    .expect("read every reconstructed byte");
+                assert_eq!(actual, payload, "the complete object must survive partial recovery");
+            }
+            for disk in set.get_disks_internal().await.into_iter().flatten() {
+                disk.close().await.expect("stop the recovered drive monitors");
+            }
        }
    }

@@ -2999,6 +2999,44 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
            let object_size = usize::try_from(fi.size)
                .map_err(|_| to_object_err(Error::other("inline fast path object size is invalid"), vec![bucket, object]))?;

+            if can_try_inline_data_shards_direct(object_size, fi.erasure.block_size)
+                && let Some(body) = verified_inline_body.filter(|body| body.len() == object_size)
+            {
+                if let Some(inline_prepare_stage_start) = inline_prepare_stage_start.take() {
+                    rustfs_io_metrics::record_get_object_stage_duration_by_size(
+                        GET_OBJECT_PATH_INLINE_DIRECT,
+                        GET_STAGE_INLINE_PREPARE,
+                        object_class.as_str(),
+                        size_bucket,
+                        inline_prepare_stage_start.elapsed().as_secs_f64(),
+                    );
+                }
+                if rustfs_io_metrics::get_stage_metrics_enabled() {
+                    rustfs_io_metrics::record_get_object_stage_duration_by_size(
+                        GET_OBJECT_PATH_INLINE_DIRECT,
+                        GET_STAGE_READER_SETUP,
+                        object_class.as_str(),
+                        size_bucket,
+                        0.0,
+                    );
+                    rustfs_io_metrics::record_get_object_stage_duration_by_size(
+                        GET_OBJECT_PATH_INLINE_DIRECT,
+                        GET_STAGE_DECODE,
+                        object_class.as_str(),
+                        size_bucket,
+                        0.0,
+                    );
+                }
+                record_get_object_reader_path_observation(GET_OBJECT_PATH_INLINE_DIRECT, object_class, size_bucket);
+                let reader = GetObjectReader {
+                    stream: Box::new(Cursor::new(body.clone())),
+                    object_info,
+                    buffered_body: Some(body),
+                    body_source: GetObjectBodySource::Unprobed,
+                };
+                return Ok(reader);
+            }
+
            let checksum_info = fi.erasure.get_checksum_info(fi.parts[0].number);
            let checksum_algo =
                if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S {
@@ -3012,42 +3050,6 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                    disks.get(index).is_some_and(Option::is_some)
                })
            {
-                if let Some(body) = verified_inline_body.filter(|body| body.len() == object_size) {
-                    if let Some(inline_prepare_stage_start) = inline_prepare_stage_start.take() {
-                        rustfs_io_metrics::record_get_object_stage_duration_by_size(
-                            GET_OBJECT_PATH_INLINE_DIRECT,
-                            GET_STAGE_INLINE_PREPARE,
-                            object_class.as_str(),
-                            size_bucket,
-                            inline_prepare_stage_start.elapsed().as_secs_f64(),
-                        );
-                    }
-                    if rustfs_io_metrics::get_stage_metrics_enabled() {
-                        rustfs_io_metrics::record_get_object_stage_duration_by_size(
-                            GET_OBJECT_PATH_INLINE_DIRECT,
-                            GET_STAGE_READER_SETUP,
-                            object_class.as_str(),
-                            size_bucket,
-                            0.0,
-                        );
-                        rustfs_io_metrics::record_get_object_stage_duration_by_size(
-                            GET_OBJECT_PATH_INLINE_DIRECT,
-                            GET_STAGE_DECODE,
-                            object_class.as_str(),
-                            size_bucket,
-                            0.0,
-                        );
-                    }
-                    record_get_object_reader_path_observation(GET_OBJECT_PATH_INLINE_DIRECT, object_class, size_bucket);
-                    let reader = GetObjectReader {
-                        stream: Box::new(Cursor::new(body.clone())),
-                        object_info,
-                        buffered_body: Some(body),
-                        body_source: GetObjectBodySource::Unprobed,
-                    };
-                    return Ok(reader);
-                }
-
                let read_length = inline_erasure_shard_file_offset(
                    0,
                    object_size,
@@ -11087,7 +11089,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
        // shared lock and permit metadata caching can early-stop once write
        // quorum agrees on the object identity.
        let allow_early_stop = !opts.no_lock || opts.metadata_cache_safe;
-        let read = self.get_object_fileinfo(bucket, object, opts, true, allow_early_stop);
+        let read = self.get_object_fileinfo_for_object_read(bucket, object, opts, true, allow_early_stop);
        #[cfg(feature = "diagnostic-read-version-caller")]
        let read = crate::cluster::rpc::runtime_sources::with_read_version_entry(
            rustfs_io_metrics::internode_metrics::ReadVersionEntrySource::ObjectInfoRead,
@@ -16904,6 +16906,7 @@ mod transition_commit_failure_tests {
                    parts_metadata,
                    online_disks,
                    read_quorum: 2,
+                    verified_inline_body: None,
                }),
            )
            .await;
@@ -22842,6 +22845,116 @@ mod put_object_tmp_cleanup_tests {
        );
    }

+    /// Per-drive view of an object: every version (including free versions)
+    /// recorded in its xl.meta, and the data directories next to it. A missing
+    /// xl.meta and one without versions both hold no version.
+    async fn per_drive_object_state(
+        dirs: &[TempDir],
+        bucket: &str,
+        object: &str,
+    ) -> Vec<(Vec<(Option<Uuid>, Option<Uuid>, Option<OffsetDateTime>)>, Vec<String>)> {
+        let mut drives = Vec::with_capacity(dirs.len());
+        for dir in dirs {
+            let object_dir = dir.path().join(bucket).join(object);
+            let versions = match tokio::fs::read(object_dir.join(STORAGE_FORMAT_FILE)).await {
+                Ok(encoded) => {
+                    let all = FileMeta::load(&encoded)
+                        .expect("xl.meta should decode")
+                        .get_all_file_info_versions(bucket, object, true)
+                        .expect("xl.meta versions should be readable");
+                    all.versions
+                        .iter()
+                        .chain(all.free_versions.iter())
+                        .map(|version| (version.version_id, version.data_dir, version.mod_time))
+                        .collect()
+                }
+                Err(err) if err.kind() == std::io::ErrorKind::NotFound => Vec::new(),
+                Err(err) => panic!("xl.meta at {object_dir:?} should be readable: {err}"),
+            };
+            let mut data_dirs = Vec::new();
+            if let Ok(mut entries) = tokio::fs::read_dir(&object_dir).await {
+                while let Some(entry) = entries.next_entry().await.expect("object directory entry should be readable") {
+                    if entry
+                        .file_type()
+                        .await
+                        .expect("object entry type should be readable")
+                        .is_dir()
+                    {
+                        data_dirs.push(entry.file_name().to_string_lossy().into_owned());
+                    }
+                }
+            }
+            data_dirs.sort();
+            drives.push((versions, data_dirs));
+        }
+        drives
+    }
+
+    /// Stage F lists any version present on half of the drives, so a PUT that
+    /// misses write quorum must not leave its version behind once its rollback
+    /// completes: no drive may keep the new xl.meta entry or its data directory.
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn failed_put_rollback_leaves_no_new_version_on_any_drive() {
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
+                for size in [4096, TEST_OBJECT_SIZE] {
+                    // (versioned, seed an earlier version first)
+                    for (versioned, seeded) in [(false, false), (false, true), (true, true)] {
+                        let case = format!("{write_completion:?} size={size} versioned={versioned} seeded={seeded}");
+                        let (dirs, disks, set) = hermetic_set_disks(4).await;
+                        let bucket = "put-rollback-no-residue";
+                        let object = "rolled-back-object";
+                        make_completion_test_bucket(&disks, bucket).await;
+                        if seeded {
+                            set.put_object(
+                                bucket,
+                                object,
+                                &mut PutObjReader::from_vec(vec![b'0'; size]),
+                                &ObjectOptions {
+                                    versioned,
+                                    write_completion: WriteCompletion::TailDrained,
+                                    ..Default::default()
+                                },
+                            )
+                            .await
+                            .unwrap_or_else(|err| panic!("{case}: seed version should commit on every drive: {err}"));
+                        }
+                        wait_for_tmp_workspace_to_drain(&dirs, "seed PUT must leave no staging").await;
+                        let before = per_drive_object_state(&dirs, bucket, object).await;
+
+                        let tasks = rename_fanout_barrier::observe_tasks(object);
+                        let _fault = rename_fault_injection::fail_rename_on(object, &[0, 1]);
+                        let err = set
+                            .put_object(
+                                bucket,
+                                object,
+                                &mut PutObjReader::from_vec(vec![b'1'; size]),
+                                &ObjectOptions {
+                                    versioned,
+                                    write_completion,
+                                    ..Default::default()
+                                },
+                            )
+                            .await
+                            .expect_err("two renamed drives cannot satisfy write quorum three");
+                        assert!(
+                            matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
+                            "{case}: original quorum error expected: {err}"
+                        );
+                        assert_eq!(tasks.running(), 0, "{case}: rename and undo tasks must be reaped before return");
+                        // An incomplete rollback retains staging for recovery, so a
+                        // drained workspace shows the rollback completed.
+                        wait_for_tmp_workspace_to_drain(&dirs, "a completed rollback must release staging").await;
+                        let after = per_drive_object_state(&dirs, bucket, object).await;
+                        assert_eq!(after, before, "{case}: no drive may keep the failed PUT's version or data");
+                    }
+                }
+            }
+        })
+        .await;
+    }
+
    #[tokio::test]
    #[serial_test::serial(capacity_dirty_scope)]
    async fn put_incomplete_rollback_preserves_staging_and_old_version_backup() {
@@ -264,6 +264,66 @@ enum MetadataReadPurpose {
    GetObjectReader,
 }

+/// How a metadata read-quorum failure is reported when the replies prove the
+/// object or version dangling. Only the object read entry points
+/// (`get_object_info`, `get_object_reader` and prepared GET metadata) report
+/// it as not found; read-before-write callers of `get_object_fileinfo*` keep
+/// the quorum error so fences and delete recovery never treat a torn write as
+/// an absent key.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum DanglingRead {
+    KeepQuorumError,
+    ReportNotFound,
+}
+
+/// Read-only form of MinIO's GET-path dangling check: once
+/// `totalResp == setDriveCount`, `getObjectFileInfo` calls `deleteIfDangling`
+/// and a dangling verdict replaces the read-quorum error with
+/// `errFileNotFound` or `errFileVersionNotFound`.
+///
+/// Each deliberate difference keeps the quorum error instead:
+/// - nothing is deleted; heal owns cleanup and its recent-write grace window;
+/// - MinIO counts an offline drive as a response and lets `isObjectDangling`
+///   reject it, while here every slot must hold valid metadata or a not-found
+///   reply, so an offline drive, a timeout, an unanswered slot, or a corrupt or
+///   unusable xl.meta stays uncertain;
+/// - the valid replies must be one metadata generation and not a remote
+///   (transitioned) version, the candidates heal's `dangling_delete_safety`
+///   also refuses to delete.
+fn dangling_read_not_found(
+    bucket: &str,
+    version_id: &str,
+    parts_metadata: &[FileInfo],
+    errs: &[Option<DiskError>],
+    fanout_complete: bool,
+) -> Option<DiskError> {
+    if !fanout_complete || crate::bucket::utils::is_meta_bucketname(bucket) {
+        return None;
+    }
+    let mut generation = None;
+    for (metadata, err) in parts_metadata.iter().zip(errs) {
+        match err {
+            Some(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {}
+            Some(_) => return None,
+            None => {
+                if !super::file_info_is_valid_for_metadata(metadata) || metadata.is_remote() {
+                    return None;
+                }
+                let identity = SetDisks::file_info_quorum_hash(metadata);
+                if *generation.get_or_insert(identity) != identity {
+                    return None;
+                }
+            }
+        }
+    }
+    let (_, dangling) = super::is_object_dangling(parts_metadata, errs, &std::collections::HashMap::new());
+    dangling.then_some(if version_id.is_empty() {
+        DiskError::FileNotFound
+    } else {
+        DiskError::FileVersionNotFound
+    })
+}
+
 impl SetDisks {
    /// A startup quorum can leave remote slots unregistered even after their
    /// peers become ready. Refresh those slots before taking the GET metadata
@@ -427,6 +487,7 @@ impl SetDisks {
        }
    }

+    #[cfg(test)]
    async fn cache_get_object_fileinfo(
        &self,
        identity: (&str, &str),
@@ -437,8 +498,32 @@ impl SetDisks {
        read_quorum: usize,
    ) {
        let (fi, capability) = selected;
+        let _ = self
+            .cache_get_object_fileinfo_owned(
+                identity,
+                generation,
+                (fi.clone(), capability, None),
+                parts_metadata.to_vec(),
+                online_disks.to_vec(),
+                read_quorum,
+            )
+            .await;
+    }
+
+    async fn cache_get_object_fileinfo_owned(
+        &self,
+        identity: (&str, &str),
+        generation: Option<GetObjectMetadataCacheGeneration>,
+        selected: (FileInfo, GetObjectMetadataCacheCapability, Option<Bytes>),
+        parts_metadata: Vec<FileInfo>,
+        online_disks: Vec<Option<DiskStore>>,
+        read_quorum: usize,
+    ) -> GetObjectFileInfo {
+        let (fi, capability, verified_inline_body) = selected;
        if fi.deleted || !fi.has_valid_erasure_geometry() {
-            return;
+            let mut snapshot = GetObjectFileInfo::owned(fi, parts_metadata, online_disks);
+            snapshot.set_verified_inline_body(verified_inline_body);
+            return snapshot;
        }
        let (bucket, object) = identity;

@@ -449,20 +534,24 @@ impl SetDisks {
                GET_METADATA_CACHE_DECISION_REJECT,
                GET_METADATA_CACHE_REASON_STALE_PUBLICATION,
            );
-            return;
+            let mut snapshot = GetObjectFileInfo::owned(fi, parts_metadata, online_disks);
+            snapshot.set_verified_inline_body(verified_inline_body);
+            return snapshot;
        };

        let key = GetObjectMetadataCacheKey::new(bucket, object, generation);
        let entry = Arc::new(GetObjectMetadataCacheEntry {
            created_at: Instant::now(),
-            fi: fi.clone(),
-            parts_metadata: parts_metadata.to_vec(),
-            online_disks: online_disks.to_vec(),
+            fi,
+            parts_metadata,
+            online_disks,
            read_quorum,
            capability,
+            verified_inline_body,
        });
-        self.insert_get_object_metadata_cache_entry_after_insert(key, generation, entry, || {})
+        self.insert_get_object_metadata_cache_entry_after_insert(key, generation, Arc::clone(&entry), || {})
            .await;
+        GetObjectFileInfo::owned_published(entry)
    }

    async fn insert_get_object_metadata_cache_entry_after_insert(
@@ -583,6 +672,31 @@ impl SetDisks {
            read_data,
            caller_allows_early_stop,
            MetadataReadPurpose::FullLayout,
+            DanglingRead::KeepQuorumError,
+        )
+        .await
+    }
+
+    /// `get_object_fileinfo` for `get_object_info` and prepared GET metadata:
+    /// a provably dangling object or version reads as not found.
+    #[tracing::instrument(level = "debug", skip(self))]
+    #[hotpath::measure(impl_type = "SetDisks")]
+    pub(super) async fn get_object_fileinfo_for_object_read(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        read_data: bool,
+        caller_allows_early_stop: bool,
+    ) -> Result<GetObjectFileInfo> {
+        self.get_object_fileinfo_gated_inner(
+            bucket,
+            object,
+            opts,
+            read_data,
+            caller_allows_early_stop,
+            MetadataReadPurpose::FullLayout,
+            DanglingRead::ReportNotFound,
        )
        .await
    }
@@ -604,6 +718,7 @@ impl SetDisks {
            read_data,
            caller_allows_early_stop,
            MetadataReadPurpose::GetObjectReader,
+            DanglingRead::ReportNotFound,
        )
        .await
    }
@@ -621,8 +736,16 @@ impl SetDisks {
        read_data: bool,
        allow_early_stop: bool,
    ) -> Result<GetObjectFileInfo> {
-        self.get_object_fileinfo_gated_inner(bucket, object, opts, read_data, allow_early_stop, MetadataReadPurpose::FullLayout)
-            .await
+        self.get_object_fileinfo_gated_inner(
+            bucket,
+            object,
+            opts,
+            read_data,
+            allow_early_stop,
+            MetadataReadPurpose::FullLayout,
+            DanglingRead::KeepQuorumError,
+        )
+        .await
    }

    #[allow(clippy::too_many_arguments)]
@@ -634,6 +757,7 @@ impl SetDisks {
        read_data: bool,
        allow_early_stop: bool,
        purpose: MetadataReadPurpose,
+        dangling_read: DanglingRead,
    ) -> Result<GetObjectFileInfo> {
        let vid = opts.version_id.clone().unwrap_or_default();
        let metadata_metrics_path = if crate::bucket::utils::is_meta_bucketname(bucket) {
@@ -793,7 +917,17 @@ impl SetDisks {
        let _min_disks = self.set_drive_count - self.default_parity_count;

        let metadata_resolve_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled);
+        // Like MinIO, check for a dangling object after either metadata quorum
+        // reduction fails; only a read-quorum failure is reclassified.
+        let dangling_not_found = |err: &DiskError| {
+            if dangling_read == DanglingRead::ReportNotFound && *err == DiskError::ErasureReadQuorum {
+                dangling_read_not_found(bucket, &vid, &parts_metadata, &errs, metadata_fanout_complete)
+            } else {
+                None
+            }
+        };
        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)
+            .map_err(|err| dangling_not_found(&err).unwrap_or(err))
            .map_err(|err| to_object_err(err.into(), vec![bucket, object, &vid]))
        {
            Ok(v) => v,
@@ -813,6 +947,15 @@ impl SetDisks {
        let write_quorum = usize::try_from(write_quorum)
            .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
        if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) {
+            if let Some(not_found) = dangling_not_found(&err) {
+                rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, "not_found");
+                record_get_stage_duration_if_enabled(
+                    GET_OBJECT_PATH_SET_DISK,
+                    GET_STAGE_METADATA_RESOLVE,
+                    metadata_resolve_stage_start,
+                );
+                return Err(to_object_err(not_found.into(), vec![bucket, object, &vid]));
+            }
            rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, "not_reached");
            error!("reduce_read_quorum_errs: {:?}, bucket: {}, object: {}", &err, bucket, object);
            record_get_stage_duration_if_enabled(
@@ -845,7 +988,11 @@ impl SetDisks {
            }
        }
        metadata_fanout_diagnostics.record_quorum_candidate_latency(metadata_metrics_path, fileinfo_selection_quorum);
-        if !opts.suppress_read_repair && errs.iter().any(|err| err.is_some()) {
+        let needs_late_metadata_fanout = !metadata_fanout_complete
+            && allow_early_stop
+            && non_inline_data_read_early_stop_allowed(read_data, bucket, object)
+            && late_materialization_candidate_is_safe(&fi);
+        let mut snapshot = if !opts.suppress_read_repair && errs.iter().any(|err| err.is_some()) {
            let version_id = resolved_read_repair_version_id(&fi, opts.version_id.as_deref());
            submit_read_repair_heal(
                &fi.volume,
@@ -857,6 +1004,9 @@ impl SetDisks {
                "metadata_read_error",
            )
            .await;
+            let mut snapshot = GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks);
+            snapshot.set_verified_inline_body(verified_inline_body);
+            snapshot
        } else if use_metadata_cache
            && (metadata_fanout_complete
                || condition_stopped
@@ -869,11 +1019,11 @@ impl SetDisks {
        {
            #[cfg(test)]
            metadata_cache_tests::wait_before_metadata_cache_publish(bucket, object).await;
-            self.cache_get_object_fileinfo(
+            self.cache_get_object_fileinfo_owned(
                (bucket, object),
                metadata_cache_generation,
                (
-                    &fi,
+                    fi,
                    if metadata_fanout_complete {
                        GetObjectMetadataCacheCapability::Full
                    } else if condition_stopped {
@@ -881,29 +1031,23 @@ impl SetDisks {
                    } else {
                        GetObjectMetadataCacheCapability::QuorumSelected
                    },
+                    verified_inline_body,
                ),
-                &parts_metadata,
-                &op_online_disks,
+                parts_metadata,
+                op_online_disks,
                read_quorum,
            )
-            .await;
-        }
-        record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_METADATA_RESOLVE, metadata_resolve_stage_start);
-        // debug!("get_object_fileinfo pick fi {:?}", &fi);
-
-        // let online_disks: Vec<Option<DiskStore>> = op_online_disks.iter().filter(|v| v.is_some()).cloned().collect();
-
-        let mut snapshot = if !metadata_fanout_complete
-            && allow_early_stop
-            && non_inline_data_read_early_stop_allowed(read_data, bucket, object)
-            && late_materialization_candidate_is_safe(&fi)
-        {
-            GetObjectFileInfo::owned_with_late_metadata_fanout(fi, parts_metadata, op_online_disks, disks)
+            .await
        } else {
-            GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks)
+            let mut snapshot = GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks);
+            snapshot.set_verified_inline_body(verified_inline_body);
+            snapshot
        };
+        record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_METADATA_RESOLVE, metadata_resolve_stage_start);
+        if needs_late_metadata_fanout {
+            snapshot.set_late_metadata_fanout_disks(disks);
+        }
        snapshot.condition_stopped = condition_stopped;
-        snapshot.set_verified_inline_body(verified_inline_body);
        Ok(snapshot)
    }

@@ -4030,6 +4174,7 @@ mod metadata_cache_tests {
                    parts_metadata: vec![fi],
                    online_disks: vec![None],
                    read_quorum: 1,
+                    verified_inline_body: None,
                }),
            )
            .await;
@@ -4691,6 +4836,7 @@ mod metadata_cache_tests {
            online_disks: Vec::new(),
            read_quorum: 0,
            capability,
+            verified_inline_body: None,
        })
    }

@@ -5183,6 +5329,7 @@ mod metadata_cache_tests {
                    parts_metadata: vec![fi],
                    online_disks: vec![None],
                    read_quorum: 1,
+                    verified_inline_body: None,
                }),
            )
            .await;
@@ -5301,6 +5448,7 @@ mod metadata_cache_tests {
            parts_metadata: vec![fi],
            online_disks: Vec::new(),
            read_quorum: 0,
+            verified_inline_body: None,
        });

        set.insert_get_object_metadata_cache_entry_after_insert(key, generation, entry, || {
@@ -5436,6 +5584,7 @@ mod metadata_cache_tests {
                fi,
                online_disks: Vec::new(),
                read_quorum: 0,
+                verified_inline_body: None,
            })
        };

@@ -9591,3 +9740,344 @@ mod tests {
        assert_eq!(errors.iter().filter(|item| item.is_none()).count(), 2);
    }
 }
+
+#[cfg(test)]
+mod dangling_read_tests {
+    //! GET and HEAD report a metadata read-quorum failure as not found only when
+    //! every drive answered and the surviving copies prove the object or version
+    //! dangling; any uncertainty keeps the read-quorum error (503 SlowDownRead).
+    use super::*;
+    use crate::disk::{DeleteOptions, STORAGE_FORMAT_FILE};
+    use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
+    use crate::storage_api_contracts::object::ObjectOperations;
+    use http::HeaderMap;
+    use std::path::PathBuf;
+    use tempfile::TempDir;
+    use tokio::io::AsyncReadExt;
+
+    /// EC 4+2: read quorum is four, so three surviving copies are below it
+    /// while still reaching the half-set probe.
+    const DRIVES: usize = 6;
+    const PARITY: usize = 2;
+
+    async fn dangling_test_set(bucket: &str) -> (Vec<TempDir>, Arc<SetDisks>) {
+        let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        ctx.update_erasure_type(crate::layout::endpoints::SetupType::Erasure).await;
+        let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(DRIVES, PARITY, ctx).await;
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("test bucket should be created on every drive");
+        (dirs, set)
+    }
+
+    async fn put(set: &SetDisks, bucket: &str, object: &str, body: &[u8], versioned: bool) -> ObjectInfo {
+        set.put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(body.to_vec()),
+            &ObjectOptions {
+                versioned,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("seed object should commit on every drive")
+    }
+
+    fn object_dir(dirs: &[TempDir], disk: usize, bucket: &str, object: &str) -> PathBuf {
+        dirs[disk].path().join(bucket).join(object)
+    }
+
+    async fn remove_object_from(dirs: &[TempDir], bucket: &str, object: &str, disks: &[usize]) {
+        for &disk in disks {
+            tokio::fs::remove_dir_all(object_dir(dirs, disk, bucket, object))
+                .await
+                .expect("object directory should be removable to model a drive that never received the write");
+        }
+    }
+
+    fn read_opts(version_id: Option<&str>) -> ObjectOptions {
+        ObjectOptions {
+            version_id: version_id.map(str::to_owned),
+            ..Default::default()
+        }
+    }
+
+    async fn head(set: &SetDisks, bucket: &str, object: &str, version_id: Option<&str>) -> Result<ObjectInfo> {
+        set.get_object_info(bucket, object, &read_opts(version_id)).await
+    }
+
+    async fn get(set: &SetDisks, bucket: &str, object: &str, version_id: Option<&str>) -> Result<Vec<u8>> {
+        let mut reader = set
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &read_opts(version_id))
+            .await?;
+        let mut body = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut body)
+            .await
+            .expect("an opened object should stream its body");
+        Ok(body)
+    }
+
+    fn is_read_quorum_error(err: &StorageError) -> bool {
+        matches!(err, StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _))
+    }
+
+    async fn assert_get_and_head_keep_read_quorum_error(set: &SetDisks, bucket: &str, object: &str, case: &str) {
+        let Err(err) = head(set, bucket, object, None).await else {
+            panic!("{case}: HEAD must not serve an object below read quorum");
+        };
+        assert!(is_read_quorum_error(&err), "{case}: HEAD must keep the read-quorum error: {err:?}");
+        let Err(err) = get(set, bucket, object, None).await else {
+            panic!("{case}: GET must not serve an object below read quorum");
+        };
+        assert!(is_read_quorum_error(&err), "{case}: GET must keep the read-quorum error: {err:?}");
+    }
+
+    #[tokio::test]
+    async fn dangling_object_reads_as_not_found_on_get_and_head() {
+        let bucket = "dangling-read-unversioned";
+        let object = "torn/object";
+        let (dirs, set) = dangling_test_set(bucket).await;
+        put(&set, bucket, object, b"torn write payload", false).await;
+        // Three copies stay below read quorum four; the other drives answer FileNotFound.
+        remove_object_from(&dirs, bucket, object, &[3, 4, 5]).await;
+
+        let is_not_found = |err: &StorageError| matches!(err, StorageError::ObjectNotFound(b, o) if b == bucket && o == object);
+        let Err(err) = head(&set, bucket, object, None).await else {
+            panic!("HEAD must not serve a dangling object");
+        };
+        assert!(is_not_found(&err), "HEAD of a dangling object must be NoSuchKey: {err:?}");
+        let Err(err) = get(&set, bucket, object, None).await else {
+            panic!("GET must not serve a dangling object");
+        };
+        assert!(is_not_found(&err), "GET of a dangling object must be NoSuchKey: {err:?}");
+        let Err(err) = set
+            .prepare_get_object_metadata(bucket, object, &ObjectOptions::default())
+            .await
+        else {
+            panic!("prepared GET metadata must not resolve a dangling object");
+        };
+        assert!(is_not_found(&err), "prepared GET metadata must be NoSuchKey: {err:?}");
+
+        // Read-before-write callers keep the quorum error so fences and delete
+        // recovery never treat a torn write as an absent key.
+        let Err(err) = set
+            .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false)
+            .await
+        else {
+            panic!("internal metadata reads must not resolve a dangling object");
+        };
+        assert!(is_read_quorum_error(&err), "internal metadata reads keep the quorum error: {err:?}");
+
+        // The classification is read-only; heal owns dangling cleanup.
+        for disk in 0..3 {
+            assert!(
+                object_dir(&dirs, disk, bucket, object).join(STORAGE_FORMAT_FILE).exists(),
+                "GET/HEAD must not delete the surviving copy on drive {disk}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn dangling_classification_requires_every_drive_to_answer() {
+        let bucket = "dangling-read-offline";
+        let object = "torn/object";
+        let (dirs, set) = dangling_test_set(bucket).await;
+        let payload = b"offline drive may hold the fourth copy";
+        put(&set, bucket, object, payload, false).await;
+        remove_object_from(&dirs, bucket, object, &[3, 4]).await;
+        let offline = set.disks.write().await[5].take().expect("drive five should start online");
+
+        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "one drive offline").await;
+
+        // The offline drive held a copy: once it answers, read quorum is met.
+        set.disks.write().await[5] = Some(offline);
+        let body = get(&set, bucket, object, None)
+            .await
+            .expect("four copies satisfy read quorum once the drive returns");
+        assert_eq!(body, payload);
+    }
+
+    #[tokio::test]
+    async fn dangling_version_reads_as_no_such_version_while_other_versions_stay_readable() {
+        let bucket = "dangling-read-versioned";
+        let object = "torn/versioned";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let v1 = put(&set, bucket, object, b"version one", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        let v2 = put(&set, bucket, object, b"version two", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        let disks = set.disks.read().await.clone();
+        for disk in disks.iter().skip(3) {
+            let disk = disk.as_ref().expect("every drive should be online");
+            let mut remove = FileInfo {
+                volume: bucket.to_string(),
+                name: object.to_string(),
+                version_id: Some(v2),
+                ..Default::default()
+            };
+            remove.set_skip_tier_free_version();
+            disk.delete_version(bucket, object, remove, false, DeleteOptions::default())
+                .await
+                .expect("dropping version two should model a drive that never received it");
+            assert!(matches!(
+                disk.read_version("", bucket, object, &v2.to_string(), &ReadOptions::default())
+                    .await,
+                Err(DiskError::FileVersionNotFound | DiskError::FileNotFound)
+            ));
+        }
+
+        let v1 = v1.to_string();
+        let v2 = v2.to_string();
+        let is_no_such_version =
+            |err: &StorageError| matches!(err, StorageError::VersionNotFound(b, o, v) if b == bucket && o == object && v == &v2);
+        let Err(err) = head(&set, bucket, object, Some(&v2)).await else {
+            panic!("HEAD must not serve a dangling version");
+        };
+        assert!(is_no_such_version(&err), "HEAD of a dangling version must be NoSuchVersion: {err:?}");
+        let Err(err) = get(&set, bucket, object, Some(&v2)).await else {
+            panic!("GET must not serve a dangling version");
+        };
+        assert!(is_no_such_version(&err), "GET of a dangling version must be NoSuchVersion: {err:?}");
+
+        assert_eq!(
+            get(&set, bucket, object, Some(&v1))
+                .await
+                .expect("the committed version must stay readable"),
+            b"version one"
+        );
+        head(&set, bucket, object, Some(&v1))
+            .await
+            .expect("HEAD of the committed version must succeed");
+        // The latest read sees two generations (version two on three drives,
+        // version one on the rest), which is not proof of absence.
+        let Err(err) = head(&set, bucket, object, None).await else {
+            panic!("a split latest version must not resolve without quorum");
+        };
+        assert!(is_read_quorum_error(&err), "a split latest version keeps the quorum error: {err:?}");
+    }
+
+    #[tokio::test]
+    async fn dangling_classification_keeps_quorum_error_for_corrupt_metadata() {
+        let bucket = "dangling-read-corrupt";
+        let object = "torn/object";
+        let (dirs, set) = dangling_test_set(bucket).await;
+        put(&set, bucket, object, b"corrupt copies are present, not absent", false).await;
+        for disk in 3..DRIVES {
+            tokio::fs::write(object_dir(&dirs, disk, bucket, object).join(STORAGE_FORMAT_FILE), b"corrupt metadata")
+                .await
+                .expect("xl.meta should be corruptible for the test");
+        }
+
+        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "corrupt xl.meta").await;
+    }
+
+    #[tokio::test]
+    async fn dangling_classification_keeps_quorum_error_for_two_generations() {
+        let bucket = "dangling-read-two-generations";
+        let object = "torn/object";
+        let (dirs, set) = dangling_test_set(bucket).await;
+        put(&set, bucket, object, b"first generation", false).await;
+        let mut first_generation = Vec::new();
+        for disk in 0..2 {
+            let path = object_dir(&dirs, disk, bucket, object).join(STORAGE_FORMAT_FILE);
+            let bytes = tokio::fs::read(&path)
+                .await
+                .expect("first generation xl.meta should be readable");
+            first_generation.push((path, bytes));
+        }
+        put(&set, bucket, object, b"second generation", false).await;
+        // First generation on two drives, second on one, three drives absent:
+        // each generation is below read quorum and neither proves the other absent.
+        for (path, bytes) in first_generation {
+            tokio::fs::write(&path, bytes)
+                .await
+                .expect("first generation xl.meta should be restorable");
+        }
+        remove_object_from(&dirs, bucket, object, &[3, 4, 5]).await;
+
+        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "two generations").await;
+    }
+
+    /// Writes `fi` as the only version on the first `copies` drives; the
+    /// remaining drives answer FileNotFound.
+    async fn write_metadata_copies(set: &SetDisks, bucket: &str, object: &str, fi: &FileInfo, copies: usize) {
+        let disks = set.disks.read().await.clone();
+        for (index, disk) in disks.iter().take(copies).enumerate() {
+            let mut copy = fi.clone();
+            copy.erasure.index = index + 1;
+            disk.as_ref()
+                .expect("every drive should be online")
+                .write_metadata(bucket, bucket, object, copy)
+                .await
+                .expect("hand-written metadata should be stored");
+        }
+    }
+
+    fn handwritten_version(bucket: &str, object: &str, data_blocks: usize, parity_blocks: usize) -> FileInfo {
+        let mut fi = FileInfo::new(object, data_blocks, parity_blocks);
+        fi.volume = bucket.to_string();
+        fi.name = object.to_string();
+        fi.size = 1;
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+        fi.metadata.insert("etag".to_string(), "handwritten-etag".to_string());
+        fi.add_object_part(1, "part-etag".to_string(), 1, None, 1, None, None);
+        fi
+    }
+
+    #[tokio::test]
+    async fn dangling_zero_parity_version_reads_as_not_found() {
+        // An EC:0 version on a default EC:2 set passes the common-parity check
+        // with four of six copies (n/2 + 1), so its read fails at the later
+        // read-quorum reduction, which must classify it as well.
+        let bucket = "dangling-read-zero-parity";
+        let object = "torn/zero-parity";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        write_metadata_copies(&set, bucket, object, &handwritten_version(bucket, object, DRIVES, 0), 4).await;
+
+        let is_not_found = |err: &StorageError| matches!(err, StorageError::ObjectNotFound(b, o) if b == bucket && o == object);
+        let Err(err) = head(&set, bucket, object, None).await else {
+            panic!("HEAD must not serve a zero-parity object missing two shards");
+        };
+        assert!(is_not_found(&err), "HEAD of a dangling zero-parity object must be NoSuchKey: {err:?}");
+        let Err(err) = get(&set, bucket, object, None).await else {
+            panic!("GET must not serve a zero-parity object missing two shards");
+        };
+        assert!(is_not_found(&err), "GET of a dangling zero-parity object must be NoSuchKey: {err:?}");
+    }
+
+    #[tokio::test]
+    async fn dangling_classification_keeps_quorum_error_for_remote_versions() {
+        // Local metadata is the only pointer to tiered data, and heal never
+        // deletes a remote candidate as dangling, so GET must not report it absent.
+        let bucket = "dangling-read-remote";
+        let object = "torn/transitioned";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let mut fi = handwritten_version(bucket, object, DRIVES - PARITY, PARITY);
+        fi.transition_status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string();
+        fi.transition_tier = "WARM".to_string();
+        fi.transitioned_objname = "remote/torn/transitioned".to_string();
+        assert!(fi.is_remote(), "the fixture must be a remote version");
+        write_metadata_copies(&set, bucket, object, &fi, 3).await;
+
+        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "remote version").await;
+    }
+
+    #[tokio::test]
+    async fn dangling_classification_skips_the_meta_bucket() {
+        // Like MinIO, internal objects are never classified: their readers
+        // treat not-found as absent configuration.
+        let object = "config/dangling-read.json";
+        let (dirs, set) = dangling_test_set("dangling-read-meta").await;
+        put(&set, RUSTFS_META_BUCKET, object, b"internal config", false).await;
+        remove_object_from(&dirs, RUSTFS_META_BUCKET, object, &[3, 4, 5]).await;
+
+        assert_get_and_head_keep_read_quorum_error(&set, RUSTFS_META_BUCKET, object, "meta bucket").await;
+    }
+}
@@ -5090,8 +5090,8 @@ mod tests {
            assert!(Arc::ptr_eq(disk, expected), "endpoint monitor must not replace an offline test handle");
            assert_eq!(disk.runtime_state(), crate::disk::health_state::RuntimeDriveHealthState::Offline);
            assert!(
-                disk.is_online().await,
-                "the valid disk identity must prevent endpoint renewal while the IO health gate remains offline"
+                !disk.is_online().await,
+                "the explicit drive fault must survive renewal despite a valid disk identity"
            );
        }
        OfflineTestDisks { disks }
@@ -14,14 +14,14 @@

 use super::*;
 use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
-use crate::disk::STORAGE_FORMAT_FILE;
+use crate::disk::{DiskAPI, STORAGE_FORMAT_FILE, WalkDirOptions};
 use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
 use bytes::Bytes;
-use rustfs_filemeta::{FileInfo, ObjectPartInfo};
+use rustfs_filemeta::{FileInfo, MetacacheReader, ObjectPartInfo};
 use std::collections::BTreeMap;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio::io::{AsyncWrite, DuplexStream};
+use tokio::io::{AsyncWrite, DuplexStream, duplex};
 use tokio::sync::Notify;
 use tokio::time::timeout;

@@ -468,6 +468,50 @@ async fn list_objects_shared_corpus_offline_corrupt_and_minority_stale() {
    fixture.assert_walk("", None, None, 3).await;
 }

+#[tokio::test]
+#[serial_test::serial]
+async fn list_objects_shared_corpus_quorum_delete_leaves_no_resurrectable_stale_copy() {
+    let mut fixture = Fixture::new("list-corpus-stale-copies", namespace(&shared_keys()), true).await;
+    // A delete that reached write quorum (three of four drives) leaves one
+    // stale copy, the most any quorum delete can leave on this set.
+    let deleted = ["a0", "z"];
+    for key in deleted {
+        let physical_key = rustfs_utils::path::encode_dir_object(key);
+        for dir in fixture.dirs.iter().take(3) {
+            tokio::fs::remove_dir_all(dir.path().join(&fixture.bucket).join(&physical_key))
+                .await
+                .expect("quorum delete fixture should remove the object directory");
+        }
+    }
+    fixture.objects.retain(|object| !deleted.contains(&object.key.as_str()));
+
+    temp_env::async_with_vars([(ENV_API_LIST_OBJECTS_QUORUM, Some("strict"))], async {
+        for max in [1, 3, 1000] {
+            fixture.assert_walk("", None, None, max).await;
+            fixture.assert_walk("", Some("/"), None, max).await;
+        }
+    })
+    .await;
+
+    // A sampling mode promises only that no committed key goes missing.
+    temp_env::async_with_vars([(ENV_API_LIST_OBJECTS_QUORUM, Some("optimal"))], async {
+        let mut listed = Vec::new();
+        let mut marker = None;
+        for _ in 0..=fixture.objects.len() + deleted.len() {
+            let page = fixture.page("", marker.clone(), None, 3).await;
+            listed.extend(page.objects.into_iter().map(|object| object.name));
+            if !page.is_truncated {
+                break;
+            }
+            marker = page.next_marker;
+        }
+        for object in &fixture.objects {
+            assert!(listed.contains(&object.key), "optimal LIST lost committed key {}", object.key);
+        }
+    })
+    .await;
+}
+
 #[tokio::test]
 async fn list_objects_shared_corpus_versions_delete_markers_and_free_versions() {
    let mut objects = vec![
@@ -446,7 +446,6 @@ fn bench_metacache_resolve(c: &mut Criterion) {
                    requested_versions: 1,
                    bucket: "bucket".to_string(),
                    strict: false,
-                    write_quorum_slack: 0,
                    candidates: Vec::new(),
                });
                black_box(resolved);
@@ -463,24 +462,6 @@ fn bench_metacache_resolve(c: &mut Criterion) {
                    requested_versions: 1,
                    bucket: "bucket".to_string(),
                    strict: false,
-                    write_quorum_slack: 0,
-                    candidates: Vec::new(),
-                });
-                black_box(resolved);
-            }
-        });
-    });
-
-    group.bench_function("8_disks_write_quorum/1000_objects", |b| {
-        b.iter(|| {
-            for entries in &sets_8_agree {
-                let resolved = entries.resolve_with_write_quorum(MetadataResolutionParams {
-                    dir_quorum: 4,
-                    obj_quorum: 4,
-                    requested_versions: 1,
-                    bucket: "bucket".to_string(),
-                    strict: true,
-                    write_quorum_slack: 0,
                    candidates: Vec::new(),
                });
                black_box(resolved);
@@ -491,13 +472,12 @@ fn bench_metacache_resolve(c: &mut Criterion) {
    group.bench_function("8_disks_disagree_merge/1000_objects", |b| {
        b.iter(|| {
            for entries in &sets_8_disagree {
-                let resolved = entries.resolve_with_write_quorum(MetadataResolutionParams {
+                let resolved = entries.resolve(MetadataResolutionParams {
                    dir_quorum: 4,
                    obj_quorum: 4,
                    requested_versions: 1,
                    bucket: "bucket".to_string(),
                    strict: true,
-                    write_quorum_slack: 0,
                    candidates: Vec::new(),
                });
                black_box(resolved);
@@ -508,13 +488,12 @@ fn bench_metacache_resolve(c: &mut Criterion) {
    group.bench_function("8_disks_agree_13_versions/1000_objects", |b| {
        b.iter(|| {
            for entries in &sets_8_agree_13v {
-                let resolved = entries.resolve_with_write_quorum(MetadataResolutionParams {
+                let resolved = entries.resolve(MetadataResolutionParams {
                    dir_quorum: 4,
                    obj_quorum: 4,
                    requested_versions: 1,
                    bucket: "bucket".to_string(),
                    strict: true,
-                    write_quorum_slack: 0,
                    candidates: Vec::new(),
                });
                black_box(resolved);
@@ -525,13 +504,12 @@ fn bench_metacache_resolve(c: &mut Criterion) {
    group.bench_function("8_disks_disagree_13_versions/1000_objects", |b| {
        b.iter(|| {
            for entries in &sets_8_disagree_13v {
-                let resolved = entries.resolve_with_write_quorum(MetadataResolutionParams {
+                let resolved = entries.resolve(MetadataResolutionParams {
                    dir_quorum: 4,
                    obj_quorum: 4,
                    requested_versions: 1,
                    bucket: "bucket".to_string(),
                    strict: true,
-                    write_quorum_slack: 0,
                    candidates: Vec::new(),
                });
                black_box(resolved);
@@ -596,191 +574,6 @@ fn bench_metacache_resolve(c: &mut Criterion) {
    group.finish();
 }

-fn listing_history(versions: usize, cached: bool) -> MetaCacheEntry {
-    let mut metadata = FileMeta::new();
-    for index in 0..versions {
-        let (data, parity) = if index + 1 == versions { (12, 4) } else { (8, 8) };
-        let mut info = FileInfo::new("object", data, parity);
-        info.volume = "bucket".to_owned();
-        info.erasure.index = 1;
-        info.data_dir = Some(Uuid::from_u128(0x1000 + u128::try_from(index).expect("data directory index")));
-        info.versioned = true;
-        info.version_id = Some(Uuid::from_u128(u128::try_from(index + 1).expect("version index")));
-        info.mod_time = Some(
-            OffsetDateTime::from_unix_timestamp(1_700_000_000 + i64::try_from(index).expect("timestamp")).expect("fixture time"),
-        );
-        info.size = 1;
-        info.parts = vec![ObjectPartInfo {
-            number: 1,
-            size: 1,
-            actual_size: 1,
-            ..Default::default()
-        }];
-        info.metadata.insert("etag".to_owned(), format!("etag-{index}"));
-        info.metadata.insert("x-amz-meta-description".to_owned(), "m".repeat(1024));
-        metadata.add_version(info).expect("valid version history");
-    }
-    MetaCacheEntry {
-        name: "object".to_owned(),
-        metadata: metadata.marshal_msg().expect("serialize history"),
-        cached: cached.then_some(metadata),
-        reusable: false,
-    }
-}
-
-#[inline]
-fn owned_agreed_latest_reference(
-    mut entry: MetaCacheEntry,
-    replicas: usize,
-    params: &rustfs_filemeta::MetadataResolutionParams,
-    enforce_write_quorum: bool,
-) -> rustfs_filemeta::MetaCacheEntryResolution {
-    use rustfs_filemeta::MetaCacheEntryResolution::{NeedsSupplement, Rejected, Resolved};
-    if !enforce_write_quorum {
-        return Resolved(entry);
-    }
-    if entry.is_dir() {
-        return if replicas >= params.dir_quorum {
-            Resolved(entry)
-        } else {
-            Rejected
-        };
-    }
-    if replicas < params.obj_quorum {
-        return Rejected;
-    }
-    let metadata = match entry.xl_meta_ref() {
-        Ok(metadata) => metadata,
-        Err(_) if params.obj_quorum <= 1 => return Resolved(entry),
-        Err(_) => return Rejected,
-    };
-    if params.requested_versions == 1
-        && let Some(first) = metadata.versions.first()
-        && !first.header.free_version()
-        && replicas >= first.listing_quorum(params.obj_quorum, params.write_quorum_slack)
-    {
-        return Resolved(entry);
-    }
-    // These fixtures cover the fast prefix, never older-version reconciliation.
-    NeedsSupplement(entry)
-}
-
-fn bench_listing_borrowed_metadata(c: &mut Criterion) {
-    use rustfs_filemeta::{FileMetaVersion, MetaCacheEntries, MetadataResolutionParams};
-    let mut group = c.benchmark_group("listing_borrowed_metadata");
-    group
-        .sample_size(20)
-        .warm_up_time(Duration::from_millis(150))
-        .measurement_time(Duration::from_millis(700));
-    for versions in [2, 16, 256] {
-        for cached in [false, true] {
-            let entry = listing_history(versions, cached);
-            let params = MetadataResolutionParams {
-                dir_quorum: 8,
-                obj_quorum: 8,
-                requested_versions: 1,
-                write_quorum_slack: 8,
-                bucket: "bucket".to_owned(),
-                ..Default::default()
-            };
-            let slots = vec![Some(&entry); 8];
-            let owned = MetaCacheEntries(vec![Some(entry.clone()); 8])
-                .resolve_with_write_quorum(params.clone())
-                .expect("previous committed version");
-            let borrowed =
-                MetaCacheEntries::resolve_borrowed_with_write_quorum(&slots, params.clone()).expect("borrowed previous version");
-            assert_eq!(owned.metadata, borrowed.metadata);
-            let id = format!("{versions}_versions/{}", if cached { "cached" } else { "cold" });
-            group.throughput(Throughput::Bytes(u64::try_from(entry.metadata.len()).expect("metadata size")));
-            // ABBA order permits an explicit baseline-drift check for each fixture.
-            for phase in ["owned_a1", "borrowed_b1", "borrowed_b2", "owned_a2"] {
-                group.bench_function(BenchmarkId::new("replicas", format!("{id}/{phase}")), |b| {
-                    b.iter(|| {
-                        if phase.starts_with("owned") {
-                            black_box(
-                                MetaCacheEntries(vec![Some(black_box(&entry).clone()); 8])
-                                    .resolve_with_write_quorum(params.clone()),
-                            )
-                        } else {
-                            let slots = vec![Some(black_box(&entry)); 8];
-                            black_box(MetaCacheEntries::resolve_borrowed_with_write_quorum(&slots, params.clone()))
-                        }
-                    });
-                });
-            }
-            let mut agreed_metadata = entry
-                .cached
-                .clone()
-                .unwrap_or_else(|| FileMeta::load(&entry.metadata).expect("agreed fixture"));
-            let mut latest = FileMetaVersion::try_from(agreed_metadata.versions[0].meta.as_slice()).expect("latest body");
-            let object = latest.object.as_mut().expect("latest object");
-            object.erasure_m = 8;
-            object.erasure_n = 8;
-            agreed_metadata.versions[0] = latest.try_into().expect("balanced latest header");
-            let agreed = MetaCacheEntry {
-                name: entry.name.clone(),
-                metadata: agreed_metadata.marshal_msg().expect("agreed history"),
-                cached: cached.then_some(agreed_metadata),
-                reusable: false,
-            };
-            for phase in ["owned_a1", "borrowed_b1", "borrowed_b2", "owned_a2"] {
-                group.bench_function(BenchmarkId::new("agreed", format!("{id}/{phase}")), |b| {
-                    b.iter_batched(
-                        || agreed.clone(),
-                        |candidate| {
-                            let resolved = if phase.starts_with("owned") {
-                                owned_agreed_latest_reference(candidate, 8, &params, true)
-                            } else {
-                                rustfs_filemeta::resolve_agreed_entry(candidate, 8, &params, true)
-                            };
-                            let rustfs_filemeta::MetaCacheEntryResolution::Resolved(candidate) = resolved else {
-                                panic!("balanced latest must use agreed fast prefix");
-                            };
-                            black_box(candidate)
-                        },
-                        BatchSize::SmallInput,
-                    );
-                });
-            }
-            let replicas: Vec<_> = (1..=8)
-                .map(|disk| {
-                    let mut metadata = entry
-                        .cached
-                        .clone()
-                        .unwrap_or_else(|| FileMeta::load(&entry.metadata).expect("decode fixture"));
-                    for version in &mut metadata.versions {
-                        let mut payload = FileMetaVersion::try_from(version.meta.as_slice()).expect("version body");
-                        payload.object.as_mut().expect("object body").erasure_index = disk;
-                        version.meta = payload.marshal_msg().expect("per-disk body");
-                    }
-                    MetaCacheEntry {
-                        name: entry.name.clone(),
-                        metadata: metadata.marshal_msg().expect("per-disk metadata"),
-                        cached: cached.then_some(metadata),
-                        reusable: false,
-                    }
-                })
-                .collect();
-            assert!(replicas[0].headers_agree(&replicas, false));
-            for phase in ["owned_a1", "borrowed_b1", "borrowed_b2", "owned_a2"] {
-                group.bench_function(BenchmarkId::new("headers", format!("{id}/{phase}")), |b| {
-                    b.iter(|| {
-                        let agreed = if phase.starts_with("owned") {
-                            replicas
-                                .iter()
-                                .all(|candidate| replicas[0].matches(Some(black_box(candidate)), false).1)
-                        } else {
-                            replicas[0].headers_agree(black_box(&replicas), false)
-                        };
-                        black_box(agreed)
-                    });
-                });
-            }
-        }
-    }
-    group.finish();
-}
-
 criterion_group!(
    benches,
    bench_create_real_xlmeta,
@@ -797,8 +590,7 @@ criterion_group!(
    bench_list_page_to_fileinfo,
    bench_delete_inline_versions,
    bench_metacache_stream_page,
-    bench_metacache_resolve,
-    bench_listing_borrowed_metadata
+    bench_metacache_resolve
 );

 criterion_main!(benches);
@@ -764,6 +764,17 @@ pub(crate) fn is_valid_distribution(distribution: &[usize], n: usize) -> bool {
 }

 impl FileInfo {
+    pub fn distribution_start(bucket: &str, object: &str, cardinality: usize) -> usize {
+        if cardinality == 0 {
+            return 0;
+        }
+        let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+        hasher.update(bucket.as_bytes());
+        hasher.update(b"/");
+        hasher.update(object.as_bytes());
+        (hasher.finalize() as u32 as usize) % cardinality
+    }
+
    pub fn new(object: &str, data_blocks: usize, parity_blocks: usize) -> Self {
        let indices = {
            let cardinality = data_blocks + parity_blocks;
@@ -115,6 +115,7 @@ pub(crate) fn data_key_for_version(version_id: Option<Uuid>) -> String {
    }
 }

+#[cfg(test)]
 fn legacy_data_key_for_version(version_id: Option<Uuid>) -> Option<String> {
    if version_id.is_none() || version_id == Some(Uuid::nil()) {
        Some(Uuid::nil().to_string())
@@ -154,7 +155,6 @@ pub fn is_skip_meta_key(key: &str) -> bool {
 }

 mod codec;
-pub(crate) use codec::{BorrowedFileMeta, reconcile_headers};
 mod inline_data;
 pub(crate) mod msgp_decode;
 mod validation;
@@ -348,19 +348,21 @@ impl FileMeta {
        self.versions.sort_by(cmp_shallow_versions_for_order);
    }

+    fn find_inline_data_slice(inline_data: &[u8], version_id: Option<Uuid>) -> Result<Option<&[u8]>> {
+        if let Some(vid) = version_id.filter(|v| !v.is_nil()) {
+            let mut buf = Uuid::encode_buffer();
+            let key = vid.hyphenated().encode_lower(&mut buf);
+            InlineData::find_slice_in_raw(inline_data, key)
+        } else {
+            if let Some(data) = InlineData::find_slice_in_raw(inline_data, NULL_VERSION_ID)? {
+                return Ok(Some(data));
+            }
+            InlineData::find_slice_in_raw(inline_data, "00000000-0000-0000-0000-000000000000")
+        }
+    }
+
    fn find_inline_data_in_slice(inline_data: &[u8], version_id: Option<Uuid>) -> Result<Option<Vec<u8>>> {
-        let key = data_key_for_version(version_id);
-        if let Some(data) = InlineData::find_in_slice(inline_data, key.as_str())? {
-            return Ok(Some(data));
-        }
-
-        if let Some(legacy_key) = legacy_data_key_for_version(version_id)
-            && legacy_key != key
-        {
-            return InlineData::find_in_slice(inline_data, legacy_key.as_str());
-        }
-
-        Ok(None)
+        Ok(Self::find_inline_data_slice(inline_data, version_id)?.map(<[u8]>::to_vec))
    }

    fn find_inline_data_for_version(&self, version_id: Option<Uuid>) -> Result<Option<Vec<u8>>> {
@@ -622,18 +622,6 @@ pub struct FileMetaShallowVersion {
    pub meta: Vec<u8>, // FileMetaVersion.marshal_msg
 }

-fn write_quorum_from_erasure(data_blocks: usize, parity_blocks: usize) -> Option<usize> {
-    if data_blocks == 0 {
-        return None;
-    }
-
-    Some(if data_blocks == parity_blocks {
-        data_blocks.saturating_add(1)
-    } else {
-        data_blocks
-    })
-}
-
 impl FileMetaShallowVersion {
    /// Parse version meta with legacy format compatibility.
    /// Use this instead of `FileMetaVersion::default()` + `unmarshal_msg()` to handle old-version xl.meta.
@@ -641,60 +629,11 @@ impl FileMetaShallowVersion {
        FileMetaVersion::try_from(self.meta.as_slice())
    }

-    pub fn write_quorum(&self, fallback_quorum: usize) -> usize {
-        write_quorum_from_header_and_meta(&self.header, self.meta.as_slice(), fallback_quorum)
-    }
-
-    /// Quorum for listing this version without accepting an unreadable payload.
-    pub fn listing_quorum(&self, fallback_quorum: usize, unreachable_disks: usize) -> usize {
-        listing_quorum_from_header_and_meta(&self.header, &self.meta, fallback_quorum, unreachable_disks)
-    }
-
    pub fn into_fileinfo(&self, volume: &str, path: &str, all_parts: bool) -> Result<FileInfo> {
        FileMetaVersion::decode_fileinfo_from_slice(self.meta.as_slice(), volume, path, all_parts, true)
    }
 }

-#[inline]
-pub(crate) fn write_quorum_from_header_and_meta(header: &FileMetaVersionHeader, meta: &[u8], fallback_quorum: usize) -> usize {
-    let header_quorum = header.write_quorum(fallback_quorum);
-    if header.has_ec() || !matches!(header.version_type, VersionType::Object | VersionType::Legacy) {
-        return header_quorum;
-    }
-
-    FileMetaVersion::try_from(meta)
-        .ok()
-        .and_then(|version| version.erasure_write_quorum())
-        .unwrap_or(header_quorum)
-}
-
-pub(crate) fn listing_quorum_from_header_and_meta(
-    header: &FileMetaVersionHeader,
-    meta: &[u8],
-    fallback_quorum: usize,
-    unreachable_disks: usize,
-) -> usize {
-    let geometry = if header.version_type == VersionType::Object && header.has_ec() {
-        Some((usize::from(header.ec_m), usize::from(header.ec_n)))
-    } else if matches!(header.version_type, VersionType::Object | VersionType::Legacy) {
-        FileMetaVersion::try_from(meta)
-            .ok()
-            .and_then(|version| version.erasure_geometry())
-    } else {
-        None
-    };
-    let Some((data_blocks, parity_blocks)) = geometry else {
-        return fallback_quorum;
-    };
-    let Some(write_quorum) = write_quorum_from_erasure(data_blocks, parity_blocks) else {
-        return fallback_quorum;
-    };
-    write_quorum
-        .saturating_sub(unreachable_disks)
-        .max(data_blocks)
-        .max(fallback_quorum)
-}
-
 impl TryFrom<FileMetaVersion> for FileMetaShallowVersion {
    type Error = Error;

@@ -737,26 +676,6 @@ pub(super) struct V2ObjectSliceSummary {
 }

 impl FileMetaVersion {
-    fn erasure_geometry(&self) -> Option<(usize, usize)> {
-        Some(match self.version_type {
-            VersionType::Object | VersionType::Legacy => {
-                if let Some(object) = &self.object {
-                    (object.erasure_m, object.erasure_n)
-                } else if let Some(object) = &self.legacy_object {
-                    (object.erasure.data_blocks, object.erasure.parity_blocks)
-                } else {
-                    return None;
-                }
-            }
-            _ => return None,
-        })
-    }
-
-    fn erasure_write_quorum(&self) -> Option<usize> {
-        let (data_blocks, parity_blocks) = self.erasure_geometry()?;
-        write_quorum_from_erasure(data_blocks, parity_blocks)
-    }
-
    /// Fast borrowed scanner for `V2Obj` slices. Returning `Ok` guarantees
    /// `FileMetaVersion::try_from` would succeed with the identical summary fields;
    /// any non-`V2Obj` or invalid input returns `Err` so callers fall back to `try_from`.
@@ -1775,14 +1694,6 @@ impl FileMetaVersionHeader {
        self.ec_m > 0 && self.ec_n > 0
    }

-    pub fn write_quorum(&self, fallback_quorum: usize) -> usize {
-        if self.version_type != VersionType::Object || !self.has_ec() {
-            return fallback_quorum;
-        }
-
-        write_quorum_from_erasure(usize::from(self.ec_m), usize::from(self.ec_n)).unwrap_or(fallback_quorum)
-    }
-
    pub fn matches_not_strict(&self, o: &FileMetaVersionHeader) -> bool {
        let mut ok = self.version_id == o.version_id && self.version_type == o.version_type && self.matches_ec(o);
        // Disk-loaded headers keep the null version as Some(nil), not None: all null
@@ -4167,15 +4078,49 @@ pub fn merge_file_meta_versions(
        .iter()
        .map(|v| v.iter().map(|ver| (&ver.header, ver.meta.as_slice())).collect())
        .collect();
-    merge_file_meta_borrowed_streams(quorum, strict, requested_versions, false, 0, streams)
+    merge_file_meta_borrowed_streams(quorum, strict, requested_versions, streams)
+}
+
+/// Picks the most common header among the tops that share `latest`'s version
+/// id; a tie keeps the header that sorts first (the newest).
+fn most_common_header_for_version_id<'a>(
+    tops: &[(&'a FileMetaVersionHeader, &'a [u8])],
+    mut latest: (&'a FileMetaVersionHeader, &'a [u8]),
+    strict: bool,
+) -> ((&'a FileMetaVersionHeader, &'a [u8]), usize) {
+    let version_id = latest.0.version_id;
+    let comparable = |header: &FileMetaVersionHeader| {
+        let mut header = header.clone();
+        if !strict {
+            header.signature = [0; 4];
+        }
+        header
+    };
+    let mut counts: HashMap<FileMetaVersionHeader, usize> = HashMap::new();
+    for &top in tops {
+        if top.0.version_id == version_id {
+            *counts.entry(comparable(top.0)).or_insert(0) += 1;
+        }
+    }
+    let mut latest_count = 0;
+    for (header, count) in &counts {
+        if *count < latest_count || (*count == latest_count && latest.0.sorts_before(header)) {
+            continue;
+        }
+        for &top in tops {
+            if comparable(top.0) == *header {
+                latest = top;
+            }
+        }
+        latest_count = *count;
+    }
+    (latest, latest_count)
 }

 pub(crate) fn merge_file_meta_borrowed_streams(
    mut quorum: usize,
    mut strict: bool,
    requested_versions: usize,
-    enforce_write_quorum: bool,
-    write_quorum_slack: usize,
    mut streams: Vec<Vec<(&FileMetaVersionHeader, &[u8])>>,
 ) -> Vec<FileMetaShallowVersion> {
    let to_shallow = |(header, meta): (&FileMetaVersionHeader, &[u8])| FileMetaShallowVersion {
@@ -4192,32 +4137,13 @@ pub(crate) fn merge_file_meta_borrowed_streams(
    }

    if streams.len() == 1 {
-        if !enforce_write_quorum {
-            return streams[0].iter().copied().map(to_shallow).collect();
-        }
-
-        let required_quorum = streams[0]
-            .first()
-            .map(|&(hdr, meta)| listing_quorum_from_header_and_meta(hdr, meta, quorum, write_quorum_slack))
-            .unwrap_or(quorum);
-        if streams.len() >= required_quorum {
-            return streams[0].iter().copied().map(to_shallow).collect();
-        }
-        return Vec::new();
+        return streams[0].iter().copied().map(to_shallow).collect();
    }

    if quorum == 1 {
        strict = true;
    }

-    let required_quorum = |(hdr, meta): (&FileMetaVersionHeader, &[u8])| {
-        if enforce_write_quorum {
-            listing_quorum_from_header_and_meta(hdr, meta, quorum, write_quorum_slack)
-        } else {
-            quorum
-        }
-    };
-
    let mut tops: Vec<(&FileMetaVersionHeader, &[u8])> = Vec::with_capacity(streams.len());

    let mut n_versions = 0;
@@ -4249,7 +4175,7 @@ pub(crate) fn merge_file_meta_borrowed_streams(
        let latest: (&FileMetaVersionHeader, &[u8]);
        if consistent {
            latest = tops[0];
-            if tops.len() >= required_quorum(latest) {
+            if tops.len() >= quorum {
                merged.push(to_shallow(latest));
                if !latest.0.free_version() {
                    n_versions += 1;
@@ -4258,6 +4184,7 @@ pub(crate) fn merge_file_meta_borrowed_streams(
        } else {
            let mut latest_ref = tops[0];
            let mut latest_count = 1;
+            let mut settled_by_version_id = false;
            for &ver in &tops[1..] {
                if ver.0 == latest_ref.0 {
                    latest_count += 1;
@@ -4281,39 +4208,26 @@ pub(crate) fn merge_file_meta_borrowed_streams(
                }

                if ver.0.version_id == latest_ref.0.version_id {
-                    let mut x: HashMap<FileMetaVersionHeader, usize> = HashMap::new();
-                    for &a in &tops {
-                        if a.0.version_id != ver.0.version_id {
-                            continue;
-                        }
-                        let mut hdr = a.0.clone();
-                        if !strict {
-                            hdr.signature = [0; 4];
-                        }
-                        *x.entry(hdr).or_insert(0) += 1;
-                    }
-                    latest_count = 0;
-                    for (k, v) in x.iter() {
-                        if *v < latest_count || (*v == latest_count && latest_ref.0.sorts_before(k)) {
-                            continue;
-                        }
-                        for &a in &tops {
-                            let mut hdr = a.0.clone();
-                            if !strict {
-                                hdr.signature = [0; 4];
-                            }
-                            if hdr == *k {
-                                latest_ref = a;
-                            }
-                        }
-
-                        latest_count = *v;
-                    }
+                    (latest_ref, latest_count) = most_common_header_for_version_id(&tops, latest_ref, strict);
+                    settled_by_version_id = true;
                    break;
                }
            }
+            // Readers return their heads in any order. When every older header
+            // sharing the newest header's version id (an interrupted null-version
+            // overwrite) came first, the scan above never compared them, and the
+            // stream trim below would then drop that older committed generation.
+            if !settled_by_version_id
+                && tops.iter().any(|ver| {
+                    ver.0 != latest_ref.0
+                        && ver.0.version_id == latest_ref.0.version_id
+                        && (strict || !ver.0.matches_not_strict(latest_ref.0))
+                })
+            {
+                (latest_ref, latest_count) = most_common_header_for_version_id(&tops, latest_ref, strict);
+            }
            latest = latest_ref;
-            if latest_count >= required_quorum(latest) {
+            if latest_count >= quorum {
                if !latest.0.free_version() {
                    n_versions += 1;
                }
@@ -4505,14 +4419,14 @@ fn read_more_sync<R: std::io::Read>(
    }

    let extra = read_size - has;
-    if buf.capacity() >= read_size {
-        // Extend the buffer if we have enough space.
-        buf.resize(read_size, 0);
-    } else {
-        buf.extend(vec![0u8; extra]);
+    buf.try_reserve(extra.saturating_add(1)).map_err(Error::other)?;
+    std::io::Read::read_to_end(&mut std::io::Read::take(reader, extra as u64), buf)?;
+    if buf.len() < read_size {
+        return Err(Error::from(std::io::Error::new(
+            std::io::ErrorKind::UnexpectedEof,
+            "failed to fill whole buffer",
+        )));
    }
-
-    reader.read_exact(&mut buf[has..])?;
    Ok(())
 }

@@ -4534,8 +4448,8 @@ pub fn read_xl_meta_no_data_sync<R: std::io::Read>(reader: &mut R, size: usize)
        has_full = false;
    }

-    let mut buf = vec![0u8; initial];
-    reader.read_exact(&mut buf)?;
+    let mut buf = Vec::new();
+    read_more_sync(reader, &mut buf, size, initial, false)?;

    let (tmp_buf, major, minor) = FileMeta::check_xl2_v1(&buf)?;

@@ -5102,39 +5016,6 @@ mod tests {
        wr
    }

-    fn encode_legacy_v2_object_body(data_blocks: usize, parity_blocks: usize) -> Vec<u8> {
-        let drive_count = data_blocks + parity_blocks;
-        let payload = LegacyObjectVersionFixture {
-            version_type: LegacyObjectVersionTypeFixture::Object,
-            object: Some(LegacyObjectFixture {
-                version_id: Some(sample_version_id().as_bytes().to_vec()),
-                data_dir: Some(Uuid::from_u128(42).as_bytes().to_vec()),
-                erasure_algorithm: "ReedSolomon".to_string(),
-                erasure_m: data_blocks,
-                erasure_n: parity_blocks,
-                erasure_block_size: 1_048_576,
-                erasure_index: 1,
-                erasure_dist: (1..=drive_count)
-                    .map(|idx| u8::try_from(idx).expect("test drive index should fit u8"))
-                    .collect(),
-                bitrot_checksum_algo: "HighwayHash".to_string(),
-                part_numbers: vec![1],
-                part_etags: vec!["etag-1".to_string()],
-                part_sizes: vec![11],
-                part_actual_sizes: vec![11],
-                part_indices: vec![Vec::new()],
-                size: 11,
-                mod_time: Some(sample_mod_time()),
-                meta_sys: HashMap::new(),
-                meta_user: HashMap::from([("content-type".to_string(), "text/plain".to_string())]),
-            }),
-            delete_marker: None,
-            write_version: 3,
-        };
-
-        rmp_serde::to_vec_named(&payload).expect("legacy object payload should marshal")
-    }
-
    #[test]
    fn version_header_unmarshal_v1_uses_legacy_layout_defaults() {
        let expected = sample_header();
@@ -5169,80 +5050,6 @@ mod tests {
        assert_eq!(decoded.ec_m, 0);
    }

-    #[test]
-    fn shallow_version_write_quorum_uses_legacy_object_payload_when_header_lacks_ec() {
-        let expected = sample_header();
-        let encoded = encode_v2_header(&expected);
-        let mut header = FileMetaVersionHeader::default();
-        header.unmarshal_v(2, &encoded).expect("legacy v2 header should decode");
-
-        let version = FileMetaShallowVersion {
-            header,
-            meta: encode_legacy_v2_object_body(7, 1),
-        };
-
-        assert_eq!(version.header.write_quorum(5), 5);
-        assert_eq!(version.write_quorum(5), 7);
-    }
-
-    #[test]
-    fn listing_quorum_uses_each_versions_data_floor() {
-        for (data, parity, offline, fallback, expected) in
-            [(8, 8, 0, 9, 9), (8, 8, 8, 8, 8), (12, 4, 8, 8, 12), (12, 4, 0, 9, 12)]
-        {
-            let version = FileMetaShallowVersion {
-                header: FileMetaVersionHeader {
-                    version_type: VersionType::Object,
-                    ec_m: data,
-                    ec_n: parity,
-                    ..Default::default()
-                },
-                // Modern headers must not decode the payload to decide quorum.
-                meta: vec![0xff],
-            };
-            assert_eq!(version.listing_quorum(fallback, offline), expected);
-        }
-        let marker = FileMetaShallowVersion {
-            header: FileMetaVersionHeader {
-                version_type: VersionType::Delete,
-                ..Default::default()
-            },
-            meta: Vec::new(),
-        };
-        assert_eq!(marker.listing_quorum(9, 0), 9, "healthy marker needs majority metadata");
-        assert_eq!(marker.listing_quorum(8, 8), 8, "degraded marker retains metadata floor");
-    }
-
-    #[test]
-    fn listing_quorum_preserves_legacy_and_zero_parity_data_floor() {
-        for (data, parity, fallback, expected) in [(12, 4, 8, 12), (8, 8, 8, 8), (4, 0, 2, 4)] {
-            let version = FileMetaShallowVersion {
-                header: FileMetaVersionHeader {
-                    version_type: VersionType::Object,
-                    ..Default::default()
-                },
-                meta: encode_legacy_v2_object_body(data, parity),
-            };
-            assert_eq!(version.listing_quorum(fallback, 8), expected);
-        }
-    }
-
-    #[test]
-    fn shallow_version_write_quorum_uses_zero_parity_object_payload() {
-        let expected = sample_header();
-        let encoded = encode_v2_header(&expected);
-        let mut header = FileMetaVersionHeader::default();
-        header.unmarshal_v(2, &encoded).expect("legacy v2 header should decode");
-
-        let version = FileMetaShallowVersion {
-            header,
-            meta: encode_legacy_v2_object_body(4, 0),
-        };
-
-        assert_eq!(version.header.write_quorum(2), 2);
-        assert_eq!(version.write_quorum(2), 4);
-    }
-
    #[test]
    fn version_header_unmarshal_v3_round_trips_current_layout() {
        let expected = sample_header();
@@ -230,7 +230,7 @@ impl InlineData {
        Self::find_in_slice(&self.0, key)
    }

-    pub(crate) fn find_in_slice(raw: &[u8], key: &str) -> Result<Option<Vec<u8>>> {
+    pub(crate) fn find_slice_in_raw<'a>(raw: &'a [u8], key: &str) -> Result<Option<&'a [u8]>> {
        if raw.is_empty() || raw[0] == 0 || raw[0] > INLINE_DATA_VER {
            return Ok(None);
        }
@@ -243,13 +243,17 @@ impl InlineData {
            fields_len -= 1;
            let (field, value) = Self::read_entry_slice(buf, &mut cur, false)?;
            if field == key {
-                return Ok(Some(value.to_vec()));
+                return Ok(Some(value));
            }
        }

        Ok(None)
    }

+    pub(crate) fn find_in_slice(raw: &[u8], key: &str) -> Result<Option<Vec<u8>>> {
+        Ok(Self::find_slice_in_raw(raw, key)?.map(<[u8]>::to_vec))
+    }
+
    pub fn validate(&self) -> Result<()> {
        Self::validate_slice(&self.0)
    }
@@ -139,15 +139,9 @@ impl KeyName {
        Self::COMMON_KEYS.iter().filter(|k| k.is_server_derived()).map(|k| k.name())
    }

+    /// Returns the policy variable form of the key, e.g. `${jwt:sub}`.
    pub fn var_name(&self) -> String {
-        match self {
-            KeyName::Aws(s) => format!("${{aws:{}}}", Into::<&str>::into(s)),
-            KeyName::Jwt(s) => format!("${{jwt:{}}}", Into::<&str>::into(s)),
-            KeyName::Ldap(s) => format!("${{ldap:{}}}", Into::<&str>::into(s)),
-            KeyName::Sts(s) => format!("${{sts:{}}}", Into::<&str>::into(s)),
-            KeyName::Svc(s) => format!("${{svc:{}}}", Into::<&str>::into(s)),
-            KeyName::S3(s) => format!("${{s3:{}}}", Into::<&str>::into(s)),
-        }
+        format!("${{{}}}", Into::<&str>::into(self))
    }
 }

@@ -414,6 +408,14 @@ mod tests {
        );
    }

+    #[test_case(KeyName::Aws(AwsKeyName::AWSSourceIP), "${aws:SourceIp}")]
+    #[test_case(KeyName::Jwt(JwtKeyName::JWTPrefUsername), "${jwt:preferred_username}")]
+    #[test_case(KeyName::Ldap(LdapKeyName::Username), "${ldap:username}")]
+    #[test_case(KeyName::S3(S3KeyName::S3LocationConstraint), "${s3:LocationConstraint}")]
+    fn var_name_matches_policy_variable_syntax(key: KeyName, expected: &str) {
+        assert_eq!(key.var_name(), expected);
+    }
+
    #[test_case("s3:x-amz-copy-source", KeyName::S3(S3KeyName::S3XAmzCopySource))]
    #[test_case("s3:VersionId", KeyName::S3(S3KeyName::S3VersionId) ; "aws_version_id")]
    #[test_case("s3:versionid", KeyName::S3(S3KeyName::S3VersionId) ; "minio_version_id")]
@@ -73,6 +73,20 @@ impl StringFunc {
    }
 }

+/// Replaces every `${...}` reference to a common condition key that has a
+/// non-empty value in the request, not only the first such key.
+fn substitute_common_keys(mut c: String, values: &HashMap<String, Vec<String>>) -> String {
+    for key in KeyName::COMMON_KEYS {
+        if let Some(v) = values.get(key.name()).and_then(|x| x.first())
+            && !v.is_empty()
+        {
+            c = c.replace(&key.var_name(), v);
+        }
+    }
+
+    c
+}
+
 impl FuncKeyValue<StringFuncValue> {
    async fn eval(
        &self,
@@ -110,17 +124,7 @@ impl FuncKeyValue<StringFuncValue> {
        let fvalues = resolved_values
            .into_iter()
            .flatten()
-            .map(|resolved_c| {
-                let mut c = Cow::from(resolved_c);
-                for key in KeyName::COMMON_KEYS {
-                    match values.get(key.name()).and_then(|x| x.first()) {
-                        Some(v) if !v.is_empty() => return Cow::Owned(c.to_mut().replace(&key.var_name(), v)),
-                        _ => continue,
-                    };
-                }
-
-                c
-            })
+            .map(|resolved_c| Cow::from(substitute_common_keys(resolved_c, values)))
            .map(|x| if ignore_case { Cow::Owned(x.to_lowercase()) } else { x })
            .collect::<Set<_>>();

@@ -159,17 +163,7 @@ impl FuncKeyValue<StringFuncValue> {
                let matched = resolved_values
                    .into_iter()
                    .flatten()
-                    .map(|resolved_c| {
-                        let mut c = Cow::from(resolved_c);
-                        for key in KeyName::COMMON_KEYS {
-                            match values.get(key.name()).and_then(|x| x.first()) {
-                                Some(v) if !v.is_empty() => return Cow::Owned(c.to_mut().replace(&key.var_name(), v)),
-                                _ => continue,
-                            };
-                        }
-
-                        c
-                    })
+                    .map(|resolved_c| substitute_common_keys(resolved_c, values))
                    .any(|x| wildcard::is_match(x, v));

                // Unqualified evaluation reports the plain aggregate match and lets the
@@ -389,6 +383,44 @@ mod tests {
        ));
    }

+    // Real requests always carry earlier common keys such as `signatureversion`
+    // and `SourceIp`; substitution must still reach later keys like `jwt:*`.
+    #[test]
+    fn test_condition_values_substitute_variables_after_other_common_keys() {
+        let request = || {
+            vec![
+                ("signatureversion", vec!["AWS4-HMAC-SHA256"]),
+                ("SourceIp", vec!["192.0.2.10"]),
+                ("preferred_username", vec!["alice"]),
+                ("prefix", vec!["alice/photos/"]),
+            ]
+        };
+
+        assert!(test_eval_like(
+            new_fkv("s3:prefix", vec!["${jwt:preferred_username}/*"]),
+            false,
+            false,
+            request()
+        ));
+        assert!(test_eval(
+            new_fkv("s3:prefix", vec!["${jwt:preferred_username}/photos/"]),
+            false,
+            false,
+            false,
+            request()
+        ));
+        assert!(!test_eval_like(
+            new_fkv("s3:prefix", vec!["${jwt:preferred_username}/*"]),
+            false,
+            false,
+            vec![
+                ("signatureversion", vec!["AWS4-HMAC-SHA256"]),
+                ("preferred_username", vec!["bob"]),
+                ("prefix", vec!["alice/photos/"]),
+            ]
+        ));
+    }
+
    #[test_case(new_fkv("s3:x-amz-copy-source", vec!["mybucket/myobject"]), false, vec![("x-amz-copy-source", vec!["mybucket/myobject"])] => true ; "1")]
    #[test_case(new_fkv("s3:x-amz-copy-source", vec!["mybucket/myobject"]), false, vec![("x-amz-copy-source", vec!["yourbucket/myobject"])] => false ; "2")]
    #[test_case(new_fkv("s3:x-amz-copy-source", vec!["mybucket/myobject"]), false, vec![] => false ; "3")]
@@ -352,6 +352,19 @@ mod tests {
        pollster::block_on(resource.is_match(object, &HashMap::new()))
    }

+    #[test_case("arn:aws:s3:::mybucket/${jwt:preferred_username}/*", "mybucket/alice/photo.jpg" => true; "jwt variable")]
+    #[test_case("arn:aws:s3:::mybucket/${jwt:preferred_username}/*", "mybucket/bob/photo.jpg" => false; "jwt variable other user")]
+    #[test_case("arn:aws:s3:::mybucket/${ldap:username}/*", "mybucket/alice-ldap/photo.jpg" => true; "ldap variable")]
+    fn test_resource_is_match_substitutes_condition_variables(resource: &str, object: &str) -> bool {
+        let resource: Resource = resource.try_into().unwrap();
+        let conditions = HashMap::from([
+            ("preferred_username".to_string(), vec!["alice".to_string()]),
+            ("username".to_string(), vec!["alice-ldap".to_string()]),
+        ]);
+
+        pollster::block_on(resource.is_match(object, &conditions))
+    }
+
    #[test]
    fn test_wildcard_resource_serialization_round_trip() {
        use crate::policy::resource::ResourceSet;
@@ -639,7 +639,7 @@ mod hot_reload_tests {
        format!("{d}BEGIN {l}{d}\n{body}\n{d}END {l}{d}\n", d = PEM_BOUNDARY_DASHES, l = PEM_OPENSSH_LABEL,)
    }

-    fn test_ed25519_pem() -> String {
+    pub(super) fn test_ed25519_pem() -> String {
        build_pem_block(
            "b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAMwAAAAtzc2gtZW\n\
             QyNTUxOQAAACCkeMEUpnJEbOMBXiQfjZcHZMEbHW3DlNRL+Jbi1cIqMgAAAKDviRiQ74kY\n\
@@ -1010,7 +1010,7 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
    fn auth_publickey_offered(
        &mut self,
        user: &str,
-        _key: &keys::PublicKey,
+        _key: russh::server::PublicKeyOffer<'_>,
    ) -> impl std::future::Future<Output = Result<Auth, Self::Error>> + Send {
        async { Ok(Auth::reject()) }
    }
@@ -1019,7 +1019,7 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
    fn auth_publickey(
        &mut self,
        user: &str,
-        _key: &keys::PublicKey,
+        _key: russh::server::VerifiedPublicKey<'_>,
    ) -> impl std::future::Future<Output = Result<Auth, Self::Error>> + Send {
        async { Ok(Auth::reject()) }
    }
@@ -1594,6 +1594,80 @@ mod tests {
        );
    }

+    #[cfg(unix)]
+    #[tokio::test]
+    async fn ssh_handshake_rejects_public_key_authentication() {
+        use crate::common::dummy_storage::DummyBackend;
+        use russh::client::{self, AuthResult};
+        use russh::keys::{PrivateKeyWithHashAlg, PublicKeyOrCertificate};
+
+        struct PinnedHostKey(keys::PublicKey);
+
+        impl client::Handler for PinnedHostKey {
+            type Error = russh::Error;
+
+            async fn check_server_key(&mut self, offered: &PublicKeyOrCertificate) -> Result<bool, Self::Error> {
+                Ok(matches!(offered, PublicKeyOrCertificate::PublicKey { key, .. } if key.key_data() == self.0.key_data()))
+            }
+        }
+
+        timeout(Duration::from_secs(30), async {
+            let key = PrivateKey::from_openssh(super::hot_reload_tests::test_ed25519_pem()).expect("decode test SSH key");
+            let trusted_host_key = key.public_key().clone();
+            let config = build_ssh_config(vec![key.clone()], 600, "SSH-2.0-RustFS");
+            let peer_addr = "127.0.0.1:12345".parse().expect("test peer address");
+            let handler = SshSessionHandler {
+                storage: Arc::new(DummyBackend::new()),
+                peer_addr,
+                session_context: None,
+                channels: HashMap::new(),
+                read_only: false,
+                part_size: super::super::test_support::TEST_PART_SIZE,
+                handles_per_session: DEFAULT_HANDLES_PER_SESSION,
+                backend_op_timeout_secs: DEFAULT_BACKEND_OP_TIMEOUT_SECS,
+                read_cache_window: READ_CACHE_WINDOW_DEFAULT,
+                read_cache_total_mem_limit: READ_CACHE_TOTAL_MEM_DEFAULT,
+                read_cache_in_use: Arc::new(AtomicU64::new(0)),
+                session_diag: Arc::new(SessionDiag::new(peer_addr, peer_addr)),
+            };
+            let (client_stream, server_stream) = tokio::io::duplex(64 * 1024);
+            let server = tokio::spawn(async move {
+                let session = russh::server::run_stream(config, server_stream, handler)
+                    .await
+                    .expect("start test SSH session");
+                // The client deliberately disconnects after the rejected authentication.
+                let _ = session.await;
+            });
+            let mut client =
+                client::connect_stream(Arc::new(client::Config::default()), client_stream, PinnedHostKey(trusted_host_key))
+                    .await
+                    .expect("complete SSH handshake with the pinned host key");
+            let auth = client
+                .authenticate_publickey("public-key-auth-disabled", PrivateKeyWithHashAlg::new(Arc::new(key), None))
+                .await
+                .expect("receive public-key authentication response");
+            match auth {
+                AuthResult::Failure {
+                    remaining_methods,
+                    partial_success,
+                } => {
+                    assert!(!partial_success, "public keys must not grant partial authentication");
+                    assert!(remaining_methods.contains(&MethodKind::Password));
+                    assert!(!remaining_methods.contains(&MethodKind::PublicKey));
+                }
+                AuthResult::Success => panic!("public-key authentication must remain disabled"),
+            }
+            client
+                .disconnect(russh::Disconnect::ByApplication, "test complete", "")
+                .await
+                .expect("disconnect test client");
+            drop(client);
+            server.await.expect("join test SSH server");
+        })
+        .await
+        .expect("SSH authentication rejection must complete within the test deadline");
+    }
+
    #[test]
    fn ssh_config_advertises_only_password() {
        let config = build_ssh_config(Vec::new(), 600, "SSH-2.0-RustFS");
@@ -0,0 +1,44 @@
+[package]
+name = "zstd"
+version = "0.13.3"
+publish = false
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+repository.workspace = true
+description = "RustFS internal adapter for zstd 0.13 consumers using the fixed codec"
+
+[dependencies]
+zstd_fixed = { package = "zstd", version = "=0.14.3", default-features = false }
+
+[features]
+default = ["legacy", "arrays", "zdict_builder"]
+arrays = ["zstd_fixed/arrays"]
+bindgen = ["zstd_fixed/bindgen"]
+debug = ["zstd_fixed/debug"]
+doc-cfg = ["zstd_fixed/doc-cfg"]
+experimental = ["zstd_fixed/experimental"]
+fat-lto = ["zstd_fixed/fat-lto"]
+legacy = ["zstd_fixed/legacy"]
+no_asm = ["zstd_fixed/no_asm"]
+pkg-config = ["zstd_fixed/pkg-config"]
+thin = ["zstd_fixed/thin"]
+thin-lto = ["zstd_fixed/thin-lto"]
+wasm = ["zstd_fixed/wasm"]
+zdict_builder = ["zstd_fixed/zdict_builder"]
+zstdmt = ["zstd_fixed/zstdmt"]
+
+[dev-dependencies]
+zstd-safe = { version = "8.2.0", features = ["seekable"] }
+toml = "0.9"
+semver = "1.0.27"
+apache-avro.workspace = true
+arrow-array = "59.3.2"
+arrow-ipc = { version = "59.3.2", features = ["zstd"] }
+arrow-schema = "59.3.2"
+parquet = { version = "59.3.2", default-features = false, features = ["arrow", "zstd"] }
+kafka-protocol = { version = "0.18.0", default-features = false, features = ["zstd"] }
+bytes.workspace = true
+
+[lints]
+workspace = true
@@ -0,0 +1,11 @@
+# Fixed zstd dependency adapter
+
+**Use this when:** maintaining the dependency bridge for consumers that still require zstd 0.13.
+
+This private package satisfies the version constraints of Avro, Arrow IPC, Parquet, and Kafka. Its version is a resolver compatibility identifier. Its entire library is a re-export of the exact registry `zstd 0.14.3` implementation, including that implementation's `zstd-safe` types. The original zstd 0.13 implementation and zstd-safe 7.x are removed from the resolved graph. Every old feature and the default feature set is forwarded.
+
+The fixed implementation closes RUSTSEC-2026-0391, RUSTSEC-2026-0398, and RUSTSEC-2026-0403. A successful advisory scan alone is insufficient evidence: cargo-deny skips local path packages. The dependency-graph test therefore requires this adapter's sole runtime dependency to be the exact fixed registry implementation and rejects vulnerable registry codec versions. Security tests exercise abandoned stream contexts, malformed legacy frames, and seekable offsets past EOF. Consumer tests retain zstandard support in all four upstream formats.
+
+Run `cargo nextest run --locked -p zstd@0.13.3` for the focused tests. The seekable feature is enabled by a test-only dependency to cover the optional vulnerable API without enabling it in production.
+
+Remove this package and the root Cargo patch when all consumers accept the fixed zstd release line directly. The cleanup marker is registered in the architecture compatibility register.
@@ -0,0 +1,20 @@
+// Copyright 2026 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.
+
+//! Internal version-constraint adapter for RustFS's locked zstd consumers.
+//!
+//! Every codec API and low-level type comes from the fixed registry implementation.
+
+// RUSTFS_COMPAT_TODO(zstd-013-fixed-backend): Bridge dependencies requiring zstd 0.13. Remove after every consumer accepts the fixed zstd line.
+pub use zstd_fixed::*;
@@ -0,0 +1,135 @@
+// Copyright 2026 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.
+
+use arrow_array::{Int32Array, RecordBatch};
+use arrow_schema::{DataType, Field, Schema};
+use std::io::Cursor;
+use std::sync::Arc;
+
+fn batch() -> RecordBatch {
+    let schema = Arc::new(Schema::new(vec![Field::new("value", DataType::Int32, false)]));
+    let values: Vec<i32> = (0..4096).map(|value| value % 7).collect();
+    RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(values))]).expect("build compressible fixture")
+}
+
+fn assert_values(actual: &RecordBatch, expected: &RecordBatch) {
+    assert_eq!(actual.num_columns(), 1);
+    let actual = actual
+        .column(0)
+        .as_any()
+        .downcast_ref::<Int32Array>()
+        .expect("decoded int32 values");
+    let expected = expected
+        .column(0)
+        .as_any()
+        .downcast_ref::<Int32Array>()
+        .expect("fixture int32 values");
+    assert_eq!(actual, expected);
+}
+
+#[test]
+fn avro_zstandard_container_preserves_records() {
+    use apache_avro::{Codec, Reader, Schema, Writer, types::Value};
+    let schema = Schema::parse_str(r#"{"type":"record","name":"Fixture","fields":[{"name":"value","type":"long"}]}"#)
+        .expect("parse Avro fixture schema");
+    let expected: Vec<_> = [0, 1, -1, i64::MAX]
+        .into_iter()
+        .map(|value| Value::Record(vec![("value".to_owned(), Value::Long(value))]))
+        .collect();
+    let mut writer =
+        Writer::with_codec(&schema, Vec::new(), Codec::Zstandard(Default::default())).expect("create Avro zstandard writer");
+    for record in &expected {
+        writer.append_value_ref(record).expect("append Avro record");
+    }
+    let encoded = writer.into_inner().expect("finish Avro zstandard container");
+    assert!(encoded.windows(b"zstandard".len()).any(|bytes| bytes == b"zstandard"));
+    let actual = Reader::new(encoded.as_slice())
+        .expect("open Avro container")
+        .collect::<Result<Vec<_>, _>>()
+        .expect("decode Avro records");
+    assert_eq!(actual, expected);
+}
+
+#[test]
+fn arrow_ipc_zstd_preserves_multiple_batches_with_reused_contexts() {
+    use arrow_ipc::{
+        CompressionType,
+        reader::FileReader,
+        writer::{FileWriter, IpcWriteOptions},
+    };
+    let expected = batch();
+    let options = IpcWriteOptions::default()
+        .try_with_compression(Some(CompressionType::ZSTD))
+        .expect("enable IPC zstd");
+    let mut encoded = Vec::new();
+    {
+        let mut writer =
+            FileWriter::try_new_with_options(&mut encoded, expected.schema().as_ref(), options).expect("open IPC writer");
+        writer.write(&expected).expect("write first compressed batch");
+        writer.write(&expected).expect("reuse IPC compression context");
+        writer.finish().expect("finish IPC file");
+    }
+    assert!(
+        encoded.windows(4).any(|bytes| bytes == [0x28, 0xb5, 0x2f, 0xfd]),
+        "fixture must contain a zstd frame"
+    );
+    let reader = FileReader::try_new(Cursor::new(encoded), None).expect("open IPC reader");
+    let actual = reader.collect::<Result<Vec<_>, _>>().expect("read compressed batches");
+    assert_eq!(actual.len(), 2);
+    for actual in actual {
+        assert_values(&actual, &expected);
+    }
+}
+
+#[test]
+fn parquet_zstd_preserves_column_values() {
+    use parquet::{
+        arrow::{ArrowWriter, arrow_reader::ParquetRecordBatchReaderBuilder},
+        basic::{Compression, ZstdLevel},
+        file::properties::WriterProperties,
+    };
+    let expected = batch();
+    let properties = WriterProperties::builder()
+        .set_compression(Compression::ZSTD(ZstdLevel::try_new(3).expect("valid level")))
+        .build();
+    let mut encoded = Vec::new();
+    {
+        let mut writer = ArrowWriter::try_new(&mut encoded, expected.schema(), Some(properties)).expect("open Parquet writer");
+        writer.write(&expected).expect("write compressed Parquet data");
+        writer.close().expect("finish Parquet file");
+    }
+    let builder = ParquetRecordBatchReaderBuilder::try_new(bytes::Bytes::from(encoded)).expect("open Parquet metadata");
+    assert!(matches!(builder.metadata().row_group(0).column(0).compression(), Compression::ZSTD(_)));
+    let mut reader = builder.with_batch_size(4096).build().expect("open Parquet reader");
+    let actual = reader.next().expect("one batch").expect("decode Parquet values");
+    assert_values(&actual, &expected);
+    assert!(reader.next().is_none());
+}
+
+#[test]
+fn kafka_zstd_preserves_record_payload() {
+    use bytes::{BufMut, Bytes, BytesMut};
+    use kafka_protocol::compression::{Compressor, Decompressor, Zstd};
+    let expected = b"Kafka record payload remains compatible".repeat(32);
+    let mut encoded = BytesMut::new();
+    <Zstd as Compressor<BytesMut>>::compress(&mut encoded, |buffer| {
+        buffer.put_slice(&expected);
+        Ok(())
+    })
+    .expect("compress Kafka records");
+    let mut encoded = encoded.freeze();
+    let actual =
+        <Zstd as Decompressor<Bytes>>::decompress(&mut encoded, |buffer| Ok(buffer.to_vec())).expect("decompress Kafka records");
+    assert_eq!(actual, expected);
+}
@@ -0,0 +1,76 @@
+// Copyright 2026 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.
+
+use semver::Version;
+use std::path::Path;
+
+#[test]
+fn legacy_constraints_resolve_only_to_the_fixed_registry_implementation() {
+    let directory = Path::new(env!("CARGO_MANIFEST_DIR"));
+    let manifest: toml::Value =
+        toml::from_str(&std::fs::read_to_string(directory.join("Cargo.toml")).expect("read adapter manifest"))
+            .expect("parse adapter manifest");
+    let dependencies = manifest["dependencies"].as_table().expect("runtime dependency table");
+    assert_eq!(dependencies.len(), 1, "the adapter must have one canonical codec implementation");
+    let implementation = &dependencies["zstd_fixed"];
+    assert_eq!(implementation["package"].as_str(), Some("zstd"));
+    assert!(implementation.get("path").is_none());
+    assert!(implementation.get("git").is_none());
+
+    let root = directory
+        .parent()
+        .expect("crates directory")
+        .parent()
+        .expect("workspace directory");
+    let root_manifest: toml::Value =
+        toml::from_str(&std::fs::read_to_string(root.join("Cargo.toml")).expect("read workspace manifest"))
+            .expect("parse workspace manifest");
+    let patch = root_manifest["patch"]["crates-io"]["zstd"]["path"]
+        .as_str()
+        .expect("local zstd patch");
+    assert_eq!(
+        root.join(patch).canonicalize().expect("resolve patch"),
+        directory.canonicalize().expect("resolve adapter")
+    );
+
+    let lock: toml::Value = toml::from_str(&std::fs::read_to_string(root.join("Cargo.lock")).expect("read workspace lockfile"))
+        .expect("parse workspace lockfile");
+    let mut adapters = 0;
+    let mut implementations = 0;
+    let mut safe_bindings = 0;
+    for package in lock["package"].as_array().expect("locked packages") {
+        let name = package["name"].as_str().expect("package name");
+        if !matches!(name, "zstd" | "zstd-safe") {
+            continue;
+        }
+        let version = Version::parse(package["version"].as_str().expect("package version")).expect("valid package semver");
+        if name == "zstd" && package.get("source").is_none() {
+            adapters += 1;
+            assert_eq!(version, Version::new(0, 13, 3));
+            let runtime = package["dependencies"].as_array().expect("adapter dependency edges");
+            assert!(runtime.iter().any(|value| value.as_str() == Some("zstd 0.14.3")));
+        } else {
+            let source = package["source"].as_str().expect("registry codec source");
+            assert!(source.starts_with("registry+"), "codec implementation must remain a registry package");
+            if name == "zstd" {
+                implementations += 1;
+                assert!(version >= Version::new(0, 14, 3), "vulnerable zstd implementation: {version}");
+            } else {
+                safe_bindings += 1;
+                assert!(version >= Version::new(8, 2, 0), "vulnerable zstd-safe implementation: {version}");
+            }
+        }
+    }
+    assert_eq!((adapters, implementations, safe_bindings), (1, 1, 1));
+}
@@ -0,0 +1,137 @@
+// Copyright 2026 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.
+
+use std::io::{Read, Write};
+use zstd::zstd_safe::{CCtx, DCtx, InBuffer, OutBuffer, ResetDirective};
+
+#[test]
+fn abandoned_encoder_cannot_prefix_the_next_request() {
+    let mut context = CCtx::create();
+    {
+        let mut abandoned = zstd::stream::Encoder::with_context(Vec::new(), &mut context);
+        abandoned
+            .write_all(b"tenant-A private input")
+            .expect("buffer the abandoned request");
+    }
+    let mut next = zstd::stream::Encoder::with_context(Vec::new(), &mut context);
+    next.write_all(b"tenant-B public input").expect("encode the next request");
+    let encoded = next.finish().expect("finish the next request");
+    assert_eq!(
+        zstd::decode_all(encoded.as_slice()).expect("decode the next frame"),
+        b"tenant-B public input"
+    );
+}
+
+#[test]
+fn abandoned_decoder_cannot_return_the_previous_request() {
+    let previous = zstd::encode_all(&b"tenant-A unread private output"[..], 1).expect("encode previous request");
+    let current = zstd::encode_all(&b"tenant-B output"[..], 1).expect("encode current request");
+    let mut context = DCtx::create();
+    {
+        let mut abandoned = zstd::stream::Decoder::with_context(previous.as_slice(), &mut context);
+        abandoned.read_exact(&mut [0; 3]).expect("partially read the previous frame");
+    }
+    let mut actual = Vec::new();
+    zstd::stream::Decoder::with_context(current.as_slice(), &mut context)
+        .read_to_end(&mut actual)
+        .expect("read the new request independently");
+    assert_eq!(actual, b"tenant-B output");
+}
+
+#[test]
+fn corrupt_legacy_frame_keeps_buffer_positions_and_length_unchanged() {
+    let corrupt = [0x27, 0xb5, 0x2f, 0xfd, 0x08, 0x00];
+    let mut context = DCtx::create();
+    let mut destination = Vec::with_capacity(4096);
+    {
+        let mut input = InBuffer::around(&corrupt);
+        let mut output = OutBuffer::around(&mut destination);
+        assert!(context.decompress_stream(&mut output, &mut input).is_err());
+        assert_eq!(input.pos(), 0);
+        assert_eq!(output.pos(), 0);
+    }
+    assert!(destination.is_empty());
+    let valid = zstd::encode_all(&b"ordinary frame after explicit reset"[..], 1).expect("encode valid control");
+    context.reset(ResetDirective::SessionOnly).expect("reset failed decoder");
+    {
+        let mut input = InBuffer::around(valid.as_slice());
+        let mut output = OutBuffer::around(&mut destination);
+        assert_eq!(context.decompress_stream(&mut output, &mut input), Ok(0));
+    }
+    assert_eq!(destination, b"ordinary frame after explicit reset");
+}
+
+#[test]
+fn corrupt_legacy_frame_cannot_emit_bytes_when_writer_is_dropped() {
+    let mut visible = Vec::new();
+    {
+        let mut decoder = zstd::stream::write::Decoder::new(&mut visible)
+            .expect("create writer decoder")
+            .auto_flush();
+        assert!(decoder.write_all(&[0x27, 0xb5, 0x2f, 0xfd, 0x08, 0x00]).is_err());
+    }
+    assert!(visible.is_empty(), "failed legacy decoding must not publish heap bytes");
+}
+
+#[test]
+fn seekable_offsets_past_eof_keep_vec_length_within_capacity() {
+    use zstd::zstd_safe::seekable::{Seekable, SeekableCStream};
+
+    let source = b"seekable frame fixture with enough data for several independent frames";
+    let mut encoded = [0; 1024];
+    let written = {
+        let mut encoder = SeekableCStream::create();
+        encoder.init(3, true, 24).expect("initialize seekable encoder");
+        let mut input = InBuffer::around(source);
+        let mut output = OutBuffer::around(&mut encoded[..]);
+        while input.pos() < source.len() {
+            let previous = input.pos();
+            encoder
+                .compress_stream(&mut output, &mut input)
+                .expect("encode seekable input");
+            assert!(input.pos() > previous, "encoder must make progress");
+        }
+        let mut finished = false;
+        for _ in 0..16 {
+            if encoder.end_stream(&mut output).expect("finish seek table") == 0 {
+                finished = true;
+                break;
+            }
+        }
+        assert!(finished, "bounded fixture must finish within the output capacity");
+        output.pos()
+    };
+    let mut decoder = Seekable::create();
+    decoder.init_buff(&encoded[..written]).expect("open seekable archive");
+    let mut decoded = Vec::with_capacity(source.len());
+    assert_eq!(decoder.decompress(&mut decoded, 0), Ok(source.len()));
+    assert_eq!(decoded, source);
+    let end = u64::try_from(source.len()).expect("fixture length fits u64");
+    for offset in [end, end + 1, end + 4096, u64::MAX - 9, u64::MAX] {
+        let mut output = Vec::with_capacity(16);
+        assert_eq!(decoder.decompress(&mut output, offset), Ok(0), "offset {offset}");
+        assert!(output.is_empty(), "offset {offset} must not extend the vector");
+        assert!(output.len() <= output.capacity());
+    }
+}
+
+#[test]
+fn stored_registry_013_frame_remains_readable() {
+    // Generated by registry zstd 0.13.3 with encode_all at level 1.
+    let frame: &[u8] = &[
+        0x28, 0xb5, 0x2f, 0xfd, 0x00, 0x48, 0x99, 0x00, 0x00, 0x73, 0x79, 0x6e, 0x74, 0x68, 0x65, 0x74, 0x69, 0x63, 0x2d, 0x72,
+        0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x2d, 0x42,
+    ];
+    assert_eq!(zstd::decode_all(frame).expect("read pre-upgrade stored frame"), b"synthetic-request-B");
+}
@@ -11,6 +11,8 @@

 ## Open Items

+- `zstd-013-fixed-backend` dependency constraint bridge: Avro, Arrow IPC, Parquet, and Kafka still require zstd 0.13. The local adapter re-exports the fixed registry zstd 0.14.3 implementation and its zstd-safe types. Remove the adapter and Cargo patch after every dependency accepts the fixed zstd line without the 0.13 bridge.
+
 - `8565` historical rebalance replacement recovery: older schema-seven intents can exhaust their retry budget before format repair when retained rebalance metadata describes removed pools. After validating blank targets, terminal historical topology, and ownership, grant one durable handoff and one rebuild attempt. Remove after supported direct upgrades no longer admit snapshots lacking replacement_preformat_handoff_retry_done and all retained affected intents have been resolved.

 - `backlog-2253-metadata-observation-adapter` metadata observations: preserve aligned metadata/error slices for existing layout, shard and exact-version PUT, MPU and delete consumers while read/heal scheduling uses typed disk observations. Pending slots acquire empty placeholders only in this adapter. Remove after those remaining consumers use typed observations; delete the adapter and old facade in a separate cleanup.
@@ -27,7 +27,7 @@ Cleanup requires `remove=true`, `dry_run=false`, an explicit non-null version, a

 Every selected disk must be online and confirm exact version absence after cleanup. Changed topology, lost fences, failed mutations, unreadable evidence, legacy markers without an incarnation, and conflicting identities produce no completion receipt. Unproven markers remain deferred with `retired_marker_proof`; a normal not-found error is not globally reclassified as success. An interrupted operation can leave partial cleanup, and replay checks the remaining actual replicas before producing an absence receipt. Deployments upgraded after an old deletion without stamps or retirement evidence retain those residual markers for separate recovery; new successful DELETE operations provide the evidence needed for automatic convergence.

-`ListObjectVersions` independently reads all reachable replicas and resolves each version with the full erasure-set metadata read quorum (`SetDisks::resolve_listed_versions` in `crates/ecstore/src/set_disk/metadata.rs`). Sample settings in `RUSTFS_LIST_OBJECTS_QUORUM` continue to govern ordinary object listing; they do not lower version-listing authority. Local reader page limits cannot truncate this discovery before the merged page is resolved. This increases metadata reads relative to sampled listing, but does not issue a separate GET for every version. Heal's union walk still discovers subquorum repair candidates.
+`ListObjectVersions` uses the listing quorum model of MinIO's `listPath` with `requestedVersions = 0` (`SetDisks::list_path` in `crates/ecstore/src/store/list_objects.rs`): `RUSTFS_LIST_OBJECTS_QUORUM` selects how many online drives are asked, and each version is kept only when at least half of the asked drives, rounded up, carry it. A version below that quorum, such as a deleted version still held by drives that were offline during the delete, is omitted from the page instead of failing it. The page still fails with a read-quorum error when fewer than half of the set's drives are online (any offline drive for a set without parity). `strict` asks every online drive; the sampled modes (`optimal`, `reduced`, `disk`) ask fewer and can briefly list a stale version that happens to fill the sample. Set-scoped scans that act on absence keep the stricter authority: `SetDisks::walk` (tier reference proofs, tier-delete journals, pool maintenance) and the recursive prefix-delete scan read every online drive and resolve each version at the read quorum of its own erasure layout (`SetDisks::resolve_listed_versions` in `crates/ecstore/src/set_disk/metadata.rs`), and a version they cannot rule out fails the scan rather than being skipped. Heal's union walk still discovers subquorum repair candidates.

 `heal_object` delegates to `heal_object_with_explicit_version_regen`, which takes the namespace write lock at entry unless `opts.no_lock` is set and binds the guard to the function scope. The guard covers the quorum metadata read, EC reconstruction, per-disk rename commit, tmp cleanup, the `HEAL_RENAME_INCOMPLETE` partial-commit return, and orphan `data_dir` reclamation (`reclaim_orphan_data_dirs`).

@@ -64,7 +64,7 @@ A single local path (`rustfs D:\rustfs\data`) is a single-node single-drive depl

 An empty `RUSTFS_STORAGE_CLASS_STANDARD` or `RUSTFS_STORAGE_CLASS_RRS` restores the automatic policy and overrides a persisted value (`standard_policy` / `rrs_policy`). `MINIO_`-prefixed spellings of these keys, of `ERASURE_SET_DRIVE_COUNT`, and of `VOLUMES` are mapped onto the `RUSTFS_` names when the `RUSTFS_` name is unset (`apply_external_env_compat` in `crates/utils/src/envs.rs`); when both are set the `RUSTFS_` value wins and the conflict is reported at startup.

-Heal knobs (`RUSTFS_HEAL_*`) and their defaults are documented once, in [scanner-runtime-controls.md](scanner-runtime-controls.md#heal-runtime-controls); the deep bitrot cycle is the admin config key `heal.bitrot_cycle` (environment `RUSTFS_SCANNER_BITROT_CYCLE_SECS`), not a `RUSTFS_HEAL_*` variable. Data-movement knobs: `RUSTFS_REBALANCE_MAX_ATTEMPTS` (default `DEFAULT_REBALANCE_MAX_ATTEMPTS` = 3, `crates/ecstore/src/services/rebalance/mod.rs`), `RUSTFS_DECOMMISSION_BUCKET_CONCURRENCY` (default cap 4) and `RUSTFS_DECOMMISSION_ENTRY_CONCURRENCY` (default cap 8, hard cap 64) in `crates/ecstore/src/core/pools.rs`, and `RUSTFS_LIST_OBJECTS_QUORUM` (default `optimal`; `disk`, `reduced`, `optimal`, `auto`, anything else `strict`, per `normalize_list_quorum` in `crates/ecstore/src/store/list_objects.rs`). None of them relaxes object quorum, object locks, or version checks.
+Heal knobs (`RUSTFS_HEAL_*`) and their defaults are documented once, in [scanner-runtime-controls.md](scanner-runtime-controls.md#heal-runtime-controls); the deep bitrot cycle is the admin config key `heal.bitrot_cycle` (environment `RUSTFS_SCANNER_BITROT_CYCLE_SECS`), not a `RUSTFS_HEAL_*` variable. Data-movement knobs: `RUSTFS_REBALANCE_MAX_ATTEMPTS` (default `DEFAULT_REBALANCE_MAX_ATTEMPTS` = 3, `crates/ecstore/src/services/rebalance/mod.rs`), `RUSTFS_DECOMMISSION_BUCKET_CONCURRENCY` (default cap 4) and `RUSTFS_DECOMMISSION_ENTRY_CONCURRENCY` (default cap 8, hard cap 64) in `crates/ecstore/src/core/pools.rs`, and `RUSTFS_LIST_OBJECTS_QUORUM` (default `strict`, which reads every online drive of the set; `disk`, `reduced`, `optimal` and `auto` read a sample, and anything else means `strict`, per `normalize_list_quorum` in `crates/ecstore/src/store/list_objects.rs`). `strict` reads all online drives, but resolves entries using a quorum of those drives. After a successful delete, a change in which drives are offline can leave enough stale copies online for the deleted key to reappear in LIST; GET or HEAD can return 404 or 503. Restore the drives and allow healing to complete before relying on degraded listings for backup, sync, replication, or lifecycle work. `optimal` samples about half the drives and can also omit a committed object whose copies mostly sit outside the sample: an `EC:8` object committed on nine of sixteen drives can miss a four-copy sample quorum. `disk` and `reduced` sample fewer drives and encounter these limits more often. Keep the default `strict` for work that needs a complete listing, while accounting for the degraded-drive limitation above. None of these settings changes object read or write quorum.

 ### 3.1 Retired listing index: upgrade and rollback

@@ -44,7 +44,7 @@ shown. The manifest contains their complete module paths.
 | MPU delete quorum | `abort_enforces_delete_write_quorum_boundary` | Abort requires the existing delete write quorum. |
 | MPU physical publication | `complete_multipart_timeout_keeps_namespace_owner_until_physical_publication` | Timeout cannot retire an active physical publication. |
 | Restore failures | `multipart_restore_aborts_every_post_create_failure` | Every injected post-create failure aborts the remote upload. |
-| Partial LIST metadata | `rustfs-filemeta`: `resolve_with_write_quorum_slack_keeps_partial_latest_hidden_during_merge` | A partial newer version remains hidden. |
+| Partial LIST metadata | `rustfs-filemeta`: `resolve_merge_keeps_the_newest_version_at_listing_quorum` | A newer version below the listing quorum remains hidden behind the committed version. |
 | Metadata observations | `metadata_observation_all_four_slot_states_and_arrival_orders_preserve_reduction` and its companion tests | Pending, absent, corrupt, offline and successful disk slots retain the existing quorum decisions across all four-slot arrival orders. |
 | Metadata cache bodies | `metadata_cache_real_get_and_range_preserve_body_across_read_modes` | Real cache misses and hits return identical full-body and non-zero Range bytes for verified inline/non-inline objects, with early-stop enabled/disabled and one offline disk. Hits share a complete four-slot snapshot containing all available disks and issue no new metadata reads. |
 | Late metadata identity | `late_metadata_refresh_rejects_changed_generation_and_mapped_shards` | Metadata from a real four-disk PUT still reaches selection quorum, but changed generation or incorrect per-slot shard mapping fails the disk-backed late-refresh boundary with an object-scoped read-quorum error. |
@@ -100,10 +100,13 @@ Removing only the two old #7049 limit guards still passes because the recursive
 child's stop result provides another protection. Removing those guards and
 ignoring that result in the same stack-draining loop makes raw limit 1 return
 `a/first` and `a0` instead of only `a/first`. This tests the historical invariant,
-not a literal revert of #7049. Setting the current offline write-quorum
-allowance to zero makes the #7010 corpus return `InsufficientReadQuorum` where
+not a literal revert of #7049. At that main, setting the offline write-quorum
+allowance to zero made the #7010 corpus return `InsufficientReadQuorum` where
 the independent oracle requires visible objects. Restoring both production
-files passes both tests. These local controls do not establish a concurrent
+files passes both tests. The listing now takes its quorum as `ceil(asked/2)`
+of the online drives it asked (MinIO's rule), so the #7010 boundary needs two
+copies among three online drives; a write-quorum requirement of three would
+hide those objects again. These local controls do not establish a concurrent
 cross-page snapshot or process-crash safety.

 ## Write owner and completion routes
@@ -396,9 +396,12 @@ fn calculate_next_marker(v2: &ListObjectsV2Output) -> Option<String> {
        .as_deref()
        .and_then(|token| decode_list_objects_v2_continuation_token(token).ok())
    {
-        // Only cache-shaped literal keys carry this server-generated escape.
-        let marker = marker.strip_suffix("[rustfs_cache:v2,return:]").unwrap_or(&marker);
-        return Some(encode_list_output_value(marker.to_owned(), v2.encoding_type.as_ref()));
+        // Keep the escape that wraps cache-shaped literal keys: a V1 client
+        // sends NextMarker back as `marker`, and the listing layer strips one
+        // legacy cursor tag from it while that compatibility window is open.
+        // Unwrapping it here would make the next page restart at the key's
+        // stem and return the same key forever.
+        return Some(encode_list_output_value(marker, v2.encoding_type.as_ref()));
    }

    let last_key = v2
@@ -584,7 +587,7 @@ mod tests {
    }

    #[test]
-    fn test_list_objects_next_marker_unwraps_only_the_literal_key_escape() {
+    fn test_list_objects_next_marker_keeps_the_literal_key_escape() {
        let key = "key[rustfs_cache:v2,return:]";
        let cursor = format!("{key}[rustfs_cache:v2,return:]");
        let v2 = ListObjectsV2Output {
@@ -593,7 +596,9 @@ mod tests {
            ..Default::default()
        };

-        assert_eq!(build_list_objects_output(v2, None).next_marker.as_deref(), Some(key));
+        // The client sends this back as `marker`; the listing layer strips the
+        // outer escape and resumes after the literal key.
+        assert_eq!(build_list_objects_output(v2, None).next_marker.as_deref(), Some(cursor.as_str()));
    }

    #[test]
Author	SHA1	Message	Date
overtrue	f83c75dc04	docs(listing): describe degraded strict quorum limits	2026-10-27 09:40:46 +08:00
Chris An	7ea3aa2a93	fix(listing): keep the V1 NextMarker escape and document optimal limits (#8976 ) A cursor for a key that literally ends in a legacy-shaped tag is issued wrapped in one extra `[rustfs_cache:v2,return:]`, because the listing layer still strips one legacy tag from incoming markers. ListObjects (V1) unwrapped it in `calculate_next_marker`; the client then sent the bare key back as `marker`, the listing layer stripped the key's own tail, and the next page restarted at the stem and returned the same key forever. Keep the wrapper in V1 NextMarker and cover the V1 resume path (marker sent back as StartAfter) on a real store. Document that sampled list quorum modes can also omit committed objects whose copies mostly sit outside the sample, and recommend `strict` for backup, sync, replication and lifecycle tooling.	2026-10-27 08:16:34 +08:00
Raphael Fakhri	4e48c390e7	fix(policy): substitute jwt and ldap variables in policies (#8927 )	2026-10-27 06:07:52 +08:00
Ramakrishna Chilaka	02030f90a3	perf(ecstore): coalesce write and multipart metadata I/O paths (#8975 ) - Coalesce InternalBuf::Ref and InternalBuf::Owned in write_all_internal onto a single blocking_queue operation, avoiding 3-5 extra spawn_blocking round-trips per disk on metadata writes. - Short-circuit recover_part_transaction when no disk reports an active part transaction (new.meta), skipping redundant rollback and part.N.meta reads on normal UploadPart calls. - Pre-allocate exact buffer capacity from file.metadata().len() in read_file_all and read_multipart_metadata, reducing per-file read() syscalls to 1 and avoiding Vec growth reallocations. - Collapse os_mkdir_all and make_dir_all to a single fast-path blocking call via create_dir_all_std. - Accept &Path in BucketExistenceCache and use cached_access in rename_file, create_file, and read_version. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a	2026-10-27 03:00:18 +08:00
Hauser	d4d416ed47	fix(deps): upgrade russh and replace vulnerable zstd codecs (#8969 ) * chore(deps): upgrade russh to 0.65.0 and refresh crates Run two cargo update and cargo upgrade rounds from origin/main. Adapt SFTP public-key callbacks to PublicKeyOffer and VerifiedPublicKey, preserving explicit rejection and password-only authentication. Cover the rejection policy with a pinned-host-key SSH handshake test. Refresh Arrow, Parquet, Brotli, ZIP, and compatible dependencies. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(deps): route legacy zstd consumers to fixed codecs Replace the old registry implementation with a private version-constraint adapter backed only by zstd 0.14.3 and zstd-safe 8.2.0. Preserve legacy features and Avro, Arrow IPC, Parquet, and Kafka formats. Cover abandoned contexts, legacy decode errors, seekable bounds, stored frames, consumer interoperability, and resolved implementation versions. Record the adapter removal condition in the compatibility register. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-10-27 02:05:58 +08:00
Hauser	fa0a056bc1	fix(ecstore): recover drives without serial reconnect delays (#8972 ) * fix(ecstore): renew faulty local drives after partial recovery Make local online detection honor the same faulty gate as foreground I/O and remote drives, allowing the existing reconnect path to replace a stranded handle once its original format is readable again. Cover partial EC:4 recovery, health-state admission, and preserve explicit test faults across handle renewal. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * perf(ecstore): reconnect set endpoints concurrently Poll independent endpoint recovery futures within the existing maximum erasure-set width so a stalled format probe cannot delay a ready drive. Keep outer-set traversal and per-slot validation and publication unchanged. Remove repetitive per-drive reconnect-loop info logs. Add an authenticated two-peer regression that holds the first format read while the second slot is admitted and serves data with its validated ID. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-10-27 02:04:46 +08:00
Chris An	1880bdca91	refactor(listing): list latest versions with MinIO's listing quorum (#8965 ) * refactor(listing): list latest versions with MinIO's listing quorum Latest-version SetDisks::list_path now follows MinIO's er.listPath: ceil(asked/2) is the listing quorum for min_disks, directories and objects, agreed entries are emitted as read, and a split sample goes through MetaCacheEntries::resolve. One object can no longer fail a page. A set with fewer than half of its drives online still fails the page with ErasureReadQuorum. Remove the write-quorum layer this replaces: ListingSupplement, ListingReconciler and the LIST object read lock, the latest_listing_* quorum helpers, resolve_agreed_entry, resolve_with_write_quorum and the enforce_write_quorum / write_quorum_slack merge branches, and the per-version listing quorum helpers in filemeta. Make merge_file_meta_borrowed_streams pick a version id's most common header regardless of reader order, so an interrupted null-version overwrite no longer drops the committed generation from the merge. * test(listing): do not assert sampled optimal listings are deterministic optimal asks a random half of the drives, so an EC:8 object committed on nine of sixteen drives can miss the four-copy listing quorum in roughly one sample in six, exactly as in MinIO. Keep the deterministic assertion for strict and for optimal when every online drive is asked.	2026-10-27 01:46:07 +08:00
Chris An	1c2f69eba9	fix(storage): 404 for dangling objects and strict list quorum default (#8963 ) * fix(storage): 404 for dangling objects and strict list quorum default Stage F.0 of the ListObjects quorum plan. Once listing returns to MinIO's listing quorum, a PUT that reached half of the drives but not read quorum becomes visible in LIST, so GET and HEAD must not answer it with a retryable 503. GET, HEAD and prepared GET metadata now report a metadata read-quorum failure as ObjectNotFound or VersionNotFound when every drive answered with valid metadata of a single generation or not-found and is_object_dangling proves the object or version dangling, matching MinIO's getObjectFileInfo. Offline or unanswered drives, timeouts, corrupt xl.meta, two generations, remote versions and the meta bucket keep the read-quorum error. Nothing is deleted on the read path; heal keeps owning cleanup and its grace window. Read-before-write callers of get_object_fileinfo keep the quorum error. RUSTFS_LIST_OBJECTS_QUORUM now defaults to strict, matching MinIO, and the operations doc explains that optimal samples drives and can briefly show deleted objects. A new test pins the write-path precondition: a PUT that misses write quorum leaves no new version or data directory on any drive once its rollback completes. * test(listing): prove the list quorum env var is read with a non-default value	2026-10-26 23:52:49 +08:00
Ramakrishna Chilaka	e193d34444	perf(ecstore): optimize cold HEAD and GET metadata read paths (#8960 ) * perf(ecstore): optimize cold HEAD and GET metadata read paths Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a * perf(ecstore): cache verified inline body and avoid early-stop allocations - Store verified_inline_body on GetObjectMetadataCacheEntry so warm GET cache hits serve the pre-verified inline Bytes payload in O(1) without re-running HighwayHash256S bitrot verification across shards. - Verify inline shard slices directly in verify_inline_body_from_observations without acquiring MODERN_REED_SOLOMON_CACHE or allocating InlineBitrotReaders. - Gate MetadataFanoutDiagnostics timers and latency sorting on get_stage_metrics_enabled(). - Avoid pinning full xl.meta buffers in GetObjectMetadataCacheEntry by using stack Uuid::encode_buffer() lookup without Bytes::slice_ref. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a * fix(ecstore): restore inline verification test context --------- Co-authored-by: Hauser <housemecn@gmail.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: overtrue <anzhengchao@gmail.com> Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a	2026-10-26 22:37:10 +08:00
Chris An	fca714efe8	fix(listing): omit sub-quorum versions from ListObjectVersions (#8964 ) Version listings bypassed listing quorum: SetDisks::list_path returned early into list_versions_authoritatively, whose resolver seeds unread slots as DiskNotFound. With any one drive of the set offline, every version held by fewer than ceil(n/2) drives (for example a deleted version kept by drives that were offline during the delete) failed the whole page with ErasureReadQuorum, surfaced as S3 503. Route merged version pages through the listing-quorum branch of SetDisks::list_path, as MinIO's listPath does with requestedVersions = 0, so such versions are omitted instead of failing the page. Keep the set-level floor: fewer than ceil(n/2) online drives still fails the page. Set walks (tier reference proofs, tier-delete journals, pool maintenance) and the recursive prefix-delete scan act on absence, so they keep the authoritative per-version scan and fail closed.	2026-10-26 22:24:20 +08:00