merge: resolve conflicts with main branch

- Merge origin/main into perf/fileinfo-optimization - Resolve conflicts in crates/ecstore/src/set_disk/ops/object.rs - Keep AHashMap import and main branch's detailed imports Co-Authored-By: heihutu <heihutu@gmail.com>
fix(s3): accept s3tables SigV4 service (#6719 )
2026-09-05 03:35:38 +00:00 · 2026-08-27 16:03:35 +08:00 · 2026-08-27 15:46:08 +08:00 · 2026-08-27 15:45:46 +08:00 · 2026-08-27 15:19:22 +08:00 · 2026-08-27 15:17:16 +08:00
54 changed files with 3402 additions and 2548 deletions
@@ -205,16 +205,6 @@ retries = 2
 filter = 'package(rustfs) & test(execute_get_object_resumes_from_relocated_pool_without_splicing_body)'
 test-group = 'ecstore-serial-flaky'

-# QUARANTINE: OPEN rustfs#6711 — the multipart fencing test's epoch helper
-# reads xl.meta back from EVERY disk, but a multipart commit only guarantees
-# quorum-many disks have persisted; a lagging disk under CI load panics the
-# read-back with "file not found" (observed on the rio-v2 leg of a
-# nextest-config-only PR; same all-disk-materialization assumption as the
-# relocated-pool fixture fixed by #6707).
-[[profile.ci.overrides]]
-filter = 'package(rustfs-ecstore) & test(object_transaction_fencing_persists_epoch_on_multipart_commit)'
-retries = 2
-
 # Serialize the 4-disk reliability / degraded-read e2e tests under the ci
 # profile too (see the e2e-reliability test-group note near the top). Not a
 # quarantine: no retries, just single-threaded so several 4-disk servers never
@@ -91,6 +91,7 @@ dependencies = [
 "const-random",
 "getrandom 0.3.4",
 "once_cell",
+ "serde",
 "version_check",
 "zerocopy",
 ]
@@ -1623,6 +1624,22 @@ dependencies = [
 "digest 0.11.3",
 ]

+[[package]]
+name = "blazesym"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "847a0a95b041ad5aae1bdc44f2bd54743f76eb0065c4f86b139f81343c287eaf"
+dependencies = [
+ "cpp_demangle",
+ "crc32fast",
+ "flate2",
+ "gimli 0.33.0",
+ "libc",
+ "memmap2",
+ "rustc-demangle",
+ "tempfile",
+]
+
 [[package]]
 name = "block-buffer"
 version = "0.10.4"
@@ -1806,6 +1823,12 @@ dependencies = [
 "libbz2-rs-sys",
 ]

+[[package]]
+name = "c-enum"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd17eb909a8c6a894926bfcc3400a4bb0e732f5a57d37b1f14e8b29e329bace8"
+
 [[package]]
 name = "camino"
 version = "1.2.5"
@@ -2300,6 +2323,15 @@ version = "0.8.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"

+[[package]]
+name = "cpp_demangle"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0667304c32ea56cb4cd6d2d7c0cfe9a2f8041229db8c033af7f8d69492429def"
+dependencies = [
+ "cfg-if",
+]
+
 [[package]]
 name = "cpubits"
 version = "0.1.1"
@@ -3453,25 +3485,23 @@ dependencies = [

 [[package]]
 name = "deadpool"
-version = "0.12.3"
+version = "0.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b"
+checksum = "3e98a7e119cd347f4201e1159b19831029e203e2d8b790547708e8157b4acf1e"
 dependencies = [
 "deadpool-runtime",
- "lazy_static",
- "num_cpus",
 "tokio",
 ]

 [[package]]
 name = "deadpool-postgres"
-version = "0.14.1"
+version = "0.14.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3d697d376cbfa018c23eb4caab1fd1883dd9c906a8c034e8d9a3cb06a7e0bef9"
+checksum = "65a536565624b97fc19f758cd01b15d12908d3344425066efc8162236fbd3749"
 dependencies = [
 "async-trait",
 "deadpool",
- "getrandom 0.2.17",
+ "getrandom 0.4.3",
 "tokio",
 "tokio-postgres",
 "tracing",
@@ -3479,9 +3509,9 @@ dependencies = [

 [[package]]
 name = "deadpool-runtime"
-version = "0.1.4"
+version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b"
+checksum = "2657f61fb1dd8bf37a8d51093cc7cee4e77125b22f7753f49b289f831bec2bae"
 dependencies = [
 "tokio",
 ]
@@ -3685,35 +3715,61 @@ dependencies = [
 ]

 [[package]]
-name = "dial9-macro"
-version = "0.3.7"
+name = "dial9-core"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1a7e31f073f2e14e5a9d338c543a0601aeaf7c43fc428cd59ce417230d0db37d"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.119",
-]
-
-[[package]]
-name = "dial9-tokio-telemetry"
-version = "0.3.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b511dfd54f5191f7eb86856fe19d8e3c8f71673bd256e1bfa1c8128fbbc0cdb"
+checksum = "9e8cbbc8955394be626249a3b52ddd6bf664373661eaeae70d87eb12bf6f20b6"
 dependencies = [
 "arc-swap",
 "bon",
 "bytes",
 "crossbeam-queue",
- "dial9-macro",
+ "dial9-trace-format",
+ "flate2",
+ "futures-util",
+ "libc",
+ "metrique",
+ "metrique-timesource",
+ "tokio",
+ "tokio-util",
+ "tracing",
+ "ulid",
+]
+
+[[package]]
+name = "dial9-perf-self-profile"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a8f65948455c504bf08576c7b5cfe93bfcaa486d661dc4ee85c2a6309ab89629"
+dependencies = [
+ "blazesym",
+ "bon",
+ "bytes",
+ "crossbeam-utils",
+ "dial9-core",
+ "dial9-trace-format",
+ "libc",
+ "perf-event-data",
+ "perf-event-open-sys2",
+ "tracing",
+]
+
+[[package]]
+name = "dial9-tokio-telemetry"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "af1244091367c805b5d98a6a590f8ab992efda06e6d2560a6967ef898ffd30a2"
+dependencies = [
+ "bon",
+ "bytes",
+ "dial9-core",
+ "dial9-perf-self-profile",
 "dial9-trace-format",
 "flate2",
 "futures-util",
 "hostname",
 "libc",
- "metrique",
 "metrique-timesource",
- "metrique-writer",
 "pin-project-lite",
 "serde",
 "serde_json",
@@ -3725,20 +3781,22 @@ dependencies = [

 [[package]]
 name = "dial9-trace-format"
-version = "0.4.1"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b3636d6ec60d94840cc414dcd6a95c77b3ba0a7b86d43fd035b89662eeb5cfa7"
+checksum = "86083b7240114b2d0da4a7e9d041571d80ca3b1f92ae0ec794f1be21709daa6a"
 dependencies = [
 "dial9-trace-format-derive",
 "serde",
+ "typeid",
 ]

 [[package]]
 name = "dial9-trace-format-derive"
-version = "0.4.1"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fff7c2855b73d0de34bc31d6dc7afbf0f6ce230a668403ac2b57b21d1ffe3928"
+checksum = "9309248f12e414d88bcc9505f78b0c9ed47f79c5b62db611493e19a612cdc9d6"
 dependencies = [
+ "proc-macro-crate",
 "proc-macro2",
 "quote",
 "syn 2.0.119",
@@ -4558,6 +4616,9 @@ version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0bf7f043f89559805f8c7cacc432749b2fa0d0a0a9ee46ce47164ed5ba7f126c"
 dependencies = [
+ "fnv",
+ "hashbrown 0.16.1",
+ "indexmap 2.14.0",
 "stable_deref_trait",
 ]

@@ -4581,20 +4642,20 @@ dependencies = [

 [[package]]
 name = "google-cloud-auth"
-version = "1.15.0"
+version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f54aab44c16b8463ae11b165a87c3d484780231f157bb1ed65843d591beb5abd"
+checksum = "ff461519b1a948200f163574be072753bcfb462a323f0eb426629d89872dd685"
 dependencies = [
 "async-trait",
 "aws-lc-rs",
- "base64 0.22.1",
+ "base64 0.23.1",
 "bytes",
- "chrono",
 "google-cloud-gax",
 "hex",
 "hmac 0.13.0",
 "http 1.5.0",
- "jsonwebtoken 10.4.0",
+ "jiff",
+ "jsonwebtoken",
 "reqwest",
 "rustc_version",
 "rustls",
@@ -4610,9 +4671,9 @@ dependencies = [

 [[package]]
 name = "google-cloud-gax"
-version = "1.13.0"
+version = "1.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9a46dd0fd026bbc4a5d84e6ab0c941cee6e3b057976a0bb107fdb5238ce598f"
+checksum = "c5615cff28ee59cfe52fbb4c11b8b1e77f650296e2ea4f4c2b7757ac6b19e752"
 dependencies = [
 "bytes",
 "futures",
@@ -4625,13 +4686,14 @@ dependencies = [
 "serde_json",
 "thiserror 2.0.20",
 "tokio",
+ "tokio-stream",
 ]

 [[package]]
 name = "google-cloud-gax-internal"
-version = "0.7.16"
+version = "0.7.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb04c54317ace06d489213f761797240b3046142a9b7ce6b9a82a9d134e193d1"
+checksum = "d2766757d877a7a8ac23da9884cb0e3f10ed9b75a0ce59801ce6b19bf9d5819e"
 dependencies = [
 "bytes",
 "futures",
@@ -4668,9 +4730,9 @@ dependencies = [

 [[package]]
 name = "google-cloud-iam-v1"
-version = "1.11.0"
+version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34cdf5acc7ef946ee2db7a7f62bd436d8395a6543b4beef110cdc061fcf578bb"
+checksum = "5f962b40234b1531e6ef73f7558871c96e117231e962086c98804323fb8d2c82"
 dependencies = [
 "async-trait",
 "bytes",
@@ -4686,9 +4748,9 @@ dependencies = [

 [[package]]
 name = "google-cloud-longrunning"
-version = "1.12.0"
+version = "1.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e6ce05df0aea2c08472983ce2bbbed9483cbb637b89ff69a7c4ef94371fe4f2"
+checksum = "1c0363c5389ffda2b55cd8a86eef4b19a3481a48467c91dc5d77f626a9572766"
 dependencies = [
 "async-trait",
 "bytes",
@@ -4704,9 +4766,9 @@ dependencies = [

 [[package]]
 name = "google-cloud-lro"
-version = "1.9.0"
+version = "1.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cd7cca2b991d619525d72a170ca7f413cb520872702442da22ac9af650a8e786"
+checksum = "47af3deef75c14a2983c430898d960c765bddbcc9f9188ca0563108e9227cfe7"
 dependencies = [
 "google-cloud-gax",
 "google-cloud-gax-internal",
@@ -4733,14 +4795,13 @@ dependencies = [

 [[package]]
 name = "google-cloud-storage"
-version = "1.17.0"
+version = "1.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9227f65175fa91a6e41f246797917697efdadfe09dd8ea84ad8b737a71efbd28"
+checksum = "973399251b245c63f1d02d0768772833dcf1372ee358f593fb9159de8fe9c7d4"
 dependencies = [
 "async-trait",
- "base64 0.22.1",
+ "base64 0.23.1",
 "bytes",
- "chrono",
 "crc32c",
 "futures",
 "google-cloud-auth",
@@ -4755,6 +4816,7 @@ dependencies = [
 "hex",
 "http 1.5.0",
 "http-body 1.1.0",
+ "jiff",
 "md5",
 "percent-encoding",
 "prost 0.14.4",
@@ -5784,22 +5846,6 @@ dependencies = [
 "wasm-bindgen",
 ]

-[[package]]
-name = "jsonwebtoken"
-version = "10.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eba32bfb4ffdeaca3e34431072faf01745c9b26d25504aa7a6cf5684334fc4fc"
-dependencies = [
- "aws-lc-rs",
- "base64 0.22.1",
- "getrandom 0.2.17",
- "js-sys",
- "serde",
- "serde_json",
- "signature 2.2.0",
- "zeroize",
-]
-
 [[package]]
 name = "jsonwebtoken"
 version = "11.0.0"
@@ -6357,7 +6403,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2b55bfa39e6f5e44a37a59a794915ce36d04371471cf62ae365cd9703f58a5e0"
 dependencies = [
 "itoa",
- "jiff",
 "metrique-core",
 "metrique-macro",
 "metrique-service-metrics",
@@ -6366,7 +6411,6 @@ dependencies = [
 "metrique-writer-core",
 "metrique-writer-macro",
 "ryu",
- "serde_json",
 "tokio",
 ]

@@ -7611,6 +7655,27 @@ version = "2.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"

+[[package]]
+name = "perf-event-data"
+version = "0.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "575828d9d7d205188048eb1508560607a03d21eafdbba47b8cade1736c1c28e1"
+dependencies = [
+ "bitflags 2.13.1",
+ "c-enum",
+ "perf-event-open-sys2",
+]
+
+[[package]]
+name = "perf-event-open-sys2"
+version = "5.0.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9c25955321465255e437600b54296983fab1feac2cd0c38958adeb26dbae49e"
+dependencies = [
+ "libc",
+ "memoffset",
+]
+
 [[package]]
 name = "petgraph"
 version = "0.7.1"
@@ -9443,7 +9508,6 @@ dependencies = [
 name = "rustfs-audit"
 version = "1.0.0-rc.4"
 dependencies = [
- "async-trait",
 "const-str",
 "futures",
 "hashbrown 0.17.1",
@@ -9537,7 +9601,7 @@ dependencies = [
 "base64-simd",
 "chacha20poly1305",
 "hotpath",
- "jsonwebtoken 11.0.0",
+ "jsonwebtoken",
 "pbkdf2 0.13.0",
 "rand 0.10.2",
 "rsa 0.10.0-rc.18",
@@ -9563,6 +9627,7 @@ dependencies = [
 name = "rustfs-ecstore"
 version = "1.0.0-rc.4"
 dependencies = [
+ "ahash",
 "arc-swap",
 "async-channel",
 "async-recursion",
@@ -9582,7 +9647,6 @@ dependencies = [
 "flatbuffers",
 "futures",
 "futures-util",
- "glob",
 "google-cloud-auth",
 "google-cloud-storage",
 "hex-simd",
@@ -9709,6 +9773,7 @@ dependencies = [
 name = "rustfs-filemeta"
 version = "1.0.0-rc.4"
 dependencies = [
+ "ahash",
 "arc-swap",
 "byteorder",
 "bytes",
@@ -9789,7 +9854,7 @@ dependencies = [
 "hmac 0.13.0",
 "hotpath",
 "http 1.5.0",
- "jsonwebtoken 11.0.0",
+ "jsonwebtoken",
 "moka",
 "openidconnect",
 "pollster",
@@ -10220,7 +10285,7 @@ dependencies = [
 "hotpath",
 "ipnetwork",
 "jiff",
- "jsonwebtoken 11.0.0",
+ "jsonwebtoken",
 "moka",
 "pollster",
 "proptest",
@@ -10441,7 +10506,6 @@ dependencies = [
 "s3s",
 "serde",
 "serde_json",
- "sha1 0.11.0",
 "sha2 0.11.0",
 "thiserror 2.0.20",
 "time",
@@ -10750,6 +10814,7 @@ dependencies = [
 name = "rustfs-utils"
 version = "1.0.0-rc.4"
 dependencies = [
+ "ahash",
 "base64-simd",
 "blake2",
 "brotli",
@@ -10981,7 +11046,7 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
 [[package]]
 name = "s3s"
 version = "0.15.0"
-source = "git+https://github.com/rustfs/s3s.git?rev=f4dedc905ec621fa85a4686df6304190b55375f6#f4dedc905ec621fa85a4686df6304190b55375f6"
+source = "git+https://github.com/rustfs/s3s.git?rev=0f6f83d98b37fd9edcaa3be573db4aa8f568e088#0f6f83d98b37fd9edcaa3be573db4aa8f568e088"
 dependencies = [
 "arc-swap",
 "arrayvec",
@@ -12813,12 +12878,28 @@ version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8e28f89b80c87b8fb0cf04ab448d5dd0dd0ade2f8891bae878de66a75a28600e"

+[[package]]
+name = "typeid"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c"
+
 [[package]]
 name = "typenum"
 version = "1.20.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"

+[[package]]
+name = "ulid"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "470dbf6591da1b39d43c14523b2b469c86879a53e8b758c8e090a470fe7b1fbe"
+dependencies = [
+ "rand 0.9.5",
+ "web-time",
+]
+
 [[package]]
 name = "unarray"
 version = "0.1.4"
@@ -12953,9 +13034,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"

 [[package]]
 name = "uuid"
-version = "1.25.0"
+version = "1.26.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f053576934f05a761a402421fbbe3d425d9366f75f978806a037b3ca481abecc"
+checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812"
 dependencies = [
 "getrandom 0.4.3",
 "js-sys",
@@ -259,8 +259,8 @@ enumset = "1.1.14"
 faster-hex = "0.10.0"
 flate2 = "1.1.9"
 glob = "0.3.4"
-google-cloud-storage = "1.17.0"
-google-cloud-auth = "1.15.0"
+google-cloud-storage = "1.18.0"
+google-cloud-auth = "1.16.0"
 hashbrown = { version = "0.17.1" }
 # Base32 for RFC 6238 TOTP shared secrets (RFC 4648 unpadded, the alphabet
 # every authenticator app expects). Already in the graph transitively.
@@ -304,7 +304,7 @@ rustify = { version = "0.7", default-features = false }
 rustix = { version = "1.1.4" }
 rust-embed = { version = "8.12.0" }
 rustc-hash = { version = "2.1.3" }
-s3s = { git = "https://github.com/rustfs/s3s.git", rev = "f4dedc905ec621fa85a4686df6304190b55375f6", version = "0.15.0", features = ["minio"] }
+s3s = { git = "https://github.com/rustfs/s3s.git", rev = "0f6f83d98b37fd9edcaa3be573db4aa8f568e088", version = "0.15.0", features = ["minio"] }
 serial_test = "4.0.1"
 shadow-rs = { default-features = false, version = "2.0.0" }
 siphasher = "1.0.3"
@@ -327,7 +327,7 @@ tracing-subscriber = { version = "0.3.23" }
 transform-stream = "0.3.1"
 url = "2.5.8"
 urlencoding = "2.1.3"
-uuid = { version = "1.25.0" }
+uuid = { version = "1.26.0" }
 vaultrs = { version = "0.8.0" }
 tar = "0.4.46"
 walkdir = "2.5.0"
@@ -341,7 +341,7 @@ zstd = "0.13.3"
 # Observability and Metrics
 metrics = "0.24.6"
 metrics-util = "0.20"
-dial9-tokio-telemetry = "0.3"
+dial9-tokio-telemetry = "0.5.0"
 opentelemetry = { version = "0.32.0" }
 opentelemetry-appender-tracing = { version = "0.32.0" }
 opentelemetry-otlp = { version = "0.32.0" }
@@ -368,6 +368,9 @@ hotpath = { version = "0.24.0", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.48" }

+# High-performance hashing
+ahash = { version = "0.8", default-features = false, features = ["std", "runtime-rng", "serde"] }
+
 [workspace.metadata.cargo-shear]
 ignored = ["hotpath", "rustfs"]

@@ -68,7 +68,6 @@ tracing = { workspace = true, features = ["std", "attributes"] }

 [dev-dependencies]
 rustfs-targets = { workspace = true, features = ["test-support"] }
-async-trait = { workspace = true }
 temp-env = { workspace = true }
 url = { workspace = true }

@@ -87,7 +87,9 @@ fn valid_config() -> PresigningConfig {
 }

 /// Flip bytes inside the `X-Amz-Signature=` query value without changing its
-/// length, producing a structurally valid but incorrect signature.
+/// length, producing a structurally valid but incorrect signature. Every hex
+/// digit is replaced by its complement (15 - v), which has no fixed point, so
+/// the tamper changes the value no matter which digits the signature contains.
 fn tamper_signature(uri: &str) -> String {
    let marker = "X-Amz-Signature=";
    let idx = uri.find(marker).expect("presigned uri must carry X-Amz-Signature") + marker.len();
@@ -96,10 +98,9 @@ fn tamper_signature(uri: &str) -> String {
    let (sig, tail) = rest.split_at(end);
    let tampered: String = sig
        .chars()
-        .map(|c| match c {
-            '0' => 'f',
-            'a' => '0',
-            other => other,
+        .map(|c| {
+            let v = c.to_digit(16).expect("X-Amz-Signature value must be hex");
+            char::from_digit(15 - v, 16).expect("complement of a hex digit is a hex digit")
        })
        .collect();
    assert_ne!(sig, tampered, "tamper must actually change the signature hex");
@@ -146,7 +146,6 @@ bytes = { workspace = true, features = ["serde"] }
 byteorder = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 jiff = { workspace = true, features = ["serde"] }
-glob = { workspace = true }
 thiserror.workspace = true
 flatbuffers.workspace = true
 futures.workspace = true
@@ -219,6 +218,9 @@ faster-hex = { workspace = true }
 ratelimit = { workspace = true }
 aws-smithy-http-client = { workspace = true, default-features = false, features = ["rustls-aws-lc"] }

+# High-performance hashing
+ahash = { workspace = true, features = ["serde"] }
+
 # Observability and Metrics
 metrics = { workspace = true }

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use ahash::AHashMap;
 use super::{metadata_boundary, object_lock_boundary, runtime_boundary as runtime_sources};
 use crate::bucket::lifecycle::bucket_lifecycle_audit::{
    LcAuditEvent, LcEventSrc, emit_non_transitioned_expiration_event, emit_transition_complete_event,
@@ -2870,7 +2871,7 @@ fn spawn_transition_transaction_recovery_once(api: Arc<ECStore>) {
 struct StaleMultipartUploadCandidate {
    path: String,
    initiated: OffsetDateTime,
-    metadata: Option<HashMap<String, String>>,
+    metadata: Option<AHashMap<String, String>>,
 }

 fn parse_stale_uploads_duration(env_key: &str, default: StdDuration) -> StdDuration {
@@ -2915,9 +2916,9 @@ async fn stale_upload_current_size(set: &Arc<SetDisks>, metadata: &HashMap<Strin
    stale_upload_current_size_with_opts(set, metadata, upload_dir, false).await
 }

-async fn stale_upload_current_size_with_opts(
+async fn stale_upload_current_size_with_opts<S: std::hash::BuildHasher>(
    set: &Arc<SetDisks>,
-    metadata: &HashMap<String, String>,
+    metadata: &HashMap<String, String, S>,
    upload_dir: &str,
    no_lock: bool,
 ) -> Option<usize> {
@@ -2950,9 +2951,9 @@ async fn stale_upload_current_size_with_opts(
    )
 }

-async fn stale_upload_lifecycle_due(
+async fn stale_upload_lifecycle_due<S: std::hash::BuildHasher>(
    set: &Arc<SetDisks>,
-    metadata: &HashMap<String, String>,
+    metadata: &HashMap<String, String, S>,
    initiated: OffsetDateTime,
    upload_dir: &str,
    no_lock: bool,
@@ -2978,7 +2979,7 @@ async fn stale_upload_lifecycle_due(
            .unwrap_or_default(),
        is_latest: true,
        delete_marker: false,
-        user_defined: metadata.clone(),
+        user_defined: metadata.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
        ..Default::default()
    };

@@ -5220,7 +5220,7 @@ mod tests {
    #[tokio::test]
    async fn server_config_snapshot_serializes_read_modify_write_transactions() {
        let baseline = encode_server_config_blob(&Config::new(), None).expect("baseline config should encode");
-        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(baseline), None));
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(baseline.clone()), None));
        let first = read_server_config_snapshot(store.clone())
            .await
            .expect("first config snapshot");
@@ -5234,7 +5234,11 @@ mod tests {
            .await
            .expect("second transaction should acquire after the first snapshot is dropped")
            .expect("second config snapshot");
-        assert!(configs_semantically_equal(&second.config, &Config::new()));
+        // Compare raw bytes against the baseline blob rather than a fresh
+        // Config::new(): the process-global DEFAULT_KVS can be registered by a
+        // sibling test mid-run, which would make a Config::new() evaluated here
+        // diverge from the baseline encoded above.
+        assert_eq!(second.raw.as_deref(), Some(baseline.as_slice()));
    }

    #[tokio::test]
@@ -5796,7 +5796,7 @@ fn decommission_remote_tiered_opts(
        versioned: version_id.is_some(),
        version_id,
        mod_time: version.mod_time,
-        user_defined: version.metadata.clone(),
+        user_defined: version.metadata.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
        src_pool_idx,
        data_movement: true,
        incl_free_versions: version.tier_free_version(),
@@ -64,7 +64,7 @@ pub(crate) const ENCRYPTED_FRAME_LAYOUT_FIXED8K_SUFFIX: &str = "encrypted-frame-
 pub(crate) const ENV_RUSTFS_ENCRYPTED_RANGE_SEEK: &str = "RUSTFS_ENCRYPTED_RANGE_SEEK";
 pub(crate) const DEFAULT_RUSTFS_ENCRYPTED_RANGE_SEEK: bool = true;

-pub(crate) fn has_encrypted_part_layout_marker(metadata: &HashMap<String, String>, suffix: &str, expected: &str) -> bool {
+pub(crate) fn has_encrypted_part_layout_marker<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>, suffix: &str, expected: &str) -> bool {
    let mut value = None;
    for (key, candidate) in metadata {
        if !rustfs_utils::http::has_internal_suffix(key, suffix) {
@@ -168,7 +168,7 @@ pub fn to_s3s_etag(etag: &str) -> ETag {
    ETag::Strong(etag.to_string())
 }

-pub fn get_raw_etag(metadata: &HashMap<String, String>) -> String {
+pub fn get_raw_etag<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>) -> String {
    metadata
        .get("etag")
        .cloned()
@@ -1008,7 +1008,7 @@ impl ObjectInfo {
            successor_mod_time: fi.successor_mod_time,
            etag,
            inlined,
-            user_defined: Arc::new(metadata),
+            user_defined: Arc::new(metadata.iter().map(|(k, v)| (k.clone(), v.clone())).collect()),
            transitioned_object,
            transition_version_state: fi.transition_version_state,
            checksum: fi.checksum.clone(),
@@ -1316,7 +1316,7 @@ impl ObjectInfo {
        if part > 0
            && let Some(checksums) = self.parts.iter().find(|p| p.number == part).and_then(|p| p.checksums.clone())
        {
-            return Ok((checksums, true));
+            return Ok((checksums.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), true));
        }

        if let Some(data) = &self.checksum {
@@ -57,7 +57,7 @@ fn rebalance_remote_tiered_opts(
        versioned: version_id.is_some(),
        version_id,
        mod_time: version.mod_time,
-        user_defined: version.metadata.clone(),
+        user_defined: version.metadata.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
        src_pool_idx,
        data_movement: true,
        include_part_checksums: true,
@@ -1454,7 +1454,7 @@ fn tier_backend_identity(config: &TierConfig) -> io::Result<TierDestinationId> {
    encode_tier_backend_identity(tier_type, endpoint, bucket, prefix, region, routing_account)
 }

-pub(crate) fn tier_destination_id_from_metadata(metadata: &HashMap<String, String>) -> io::Result<Option<TierDestinationId>> {
+pub(crate) fn tier_destination_id_from_metadata<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>) -> io::Result<Option<TierDestinationId>> {
    let Some(encoded) = rustfs_utils::http::metadata_compat::get_consistent_str(
        metadata,
        rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
@@ -609,7 +609,7 @@ impl SetDisks {
            || Self::starts_with_ignore_ascii_case(suffix, http::SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX)
    }

-    fn update_hash_quorum_metadata_map(hasher: &mut Sha256, entries: &HashMap<String, String>) {
+    fn update_hash_quorum_metadata_map<S: std::hash::BuildHasher>(hasher: &mut Sha256, entries: &HashMap<String, String, S>) {
        let mut entries = entries
            .iter()
            .filter(|(name, _)| !Self::is_replication_quorum_metadata_key(name))
@@ -635,7 +635,7 @@ impl SetDisks {
    /// so the dual internal prefixes carrying the same mapping share one
    /// identity, while a genuine disagreement between disks still changes the
    /// hash and surfaces as a quorum difference.
-    fn update_hash_target_delete_marker_versions(hasher: &mut Sha256, metadata: &HashMap<String, String>) {
+    fn update_hash_target_delete_marker_versions<S: std::hash::BuildHasher>(hasher: &mut Sha256, metadata: &HashMap<String, String, S>) {
        let (versions, corrupt) = http::target_delete_marker_versions(metadata);
        hasher.update([u8::from(corrupt)]);
        let mut versions = versions.iter().collect::<Vec<_>>();
@@ -190,7 +190,7 @@ use tracing::error;
 use tracing::{Instrument, debug, info, warn};
 use uuid::Uuid;

-pub(super) fn restore_operation_id_from_metadata(metadata: &HashMap<String, String>) -> Result<Option<Uuid>> {
+pub(super) fn restore_operation_id_from_metadata<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>) -> Result<Option<Uuid>> {
    let Some(value) = rustfs_utils::http::metadata_compat::get_consistent_str(metadata, SUFFIX_RESTORE_OPERATION_ID) else {
        if rustfs_utils::http::metadata_compat::contains_key_str(metadata, SUFFIX_RESTORE_OPERATION_ID) {
            return Err(Error::other("invalid restore operation id metadata".to_string()));
@@ -204,14 +204,14 @@ pub(super) fn restore_operation_id_from_metadata(metadata: &HashMap<String, Stri
    Ok(Some(id))
 }

-pub(super) fn require_restore_operation_id(metadata: &HashMap<String, String>, expected: Uuid) -> Result<()> {
+pub(super) fn require_restore_operation_id<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>, expected: Uuid) -> Result<()> {
    match restore_operation_id_from_metadata(metadata)? {
        Some(actual) if actual == expected => Ok(()),
        _ => Err(Error::other("restore operation id changed before copy-back".to_string())),
    }
 }

-pub(super) fn restore_commit_operation_id_from_metadata(metadata: &HashMap<String, String>) -> Result<Option<Uuid>> {
+pub(super) fn restore_commit_operation_id_from_metadata<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>) -> Result<Option<Uuid>> {
    if !metadata.contains_key(X_AMZ_RESTORE.as_str()) {
        return Ok(None);
    }
@@ -466,13 +466,13 @@ fn release_materialized_read_lock(bucket: &str, object: &str, read_lock_guard: O
    drop(read_lock_guard);
 }

-pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap<String, String>) {
+pub(crate) fn strip_internal_multipart_metadata<S: std::hash::BuildHasher>(metadata: &mut HashMap<String, String, S>) {
    metadata.remove(RUSTFS_MULTIPART_BUCKET_KEY);
    metadata.remove(RUSTFS_MULTIPART_OBJECT_KEY);
    rustfs_utils::http::metadata_compat::remove_str(metadata, SUFFIX_BUCKET_INCARNATION_ID);
 }

-fn should_persist_encryption_original_size(metadata: &HashMap<String, String>) -> bool {
+fn should_persist_encryption_original_size<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>) -> bool {
    metadata.keys().any(|key| is_object_encryption_marker(key))
 }

@@ -365,7 +365,7 @@ fn fence_commit_on_lock_loss(guard: Option<&ObjectLockDiagGuard>, mode: &'static
    Ok(())
 }

-fn multipart_bucket_incarnation_id(metadata: &HashMap<String, String>) -> Result<Option<Uuid>> {
+fn multipart_bucket_incarnation_id<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>) -> Result<Option<Uuid>> {
    let Some(value) = rustfs_utils::http::metadata_compat::get_consistent_str(metadata, SUFFIX_BUCKET_INCARNATION_ID) else {
        if rustfs_utils::http::metadata_compat::contains_key_str(metadata, SUFFIX_BUCKET_INCARNATION_ID) {
            return Err(Error::other("invalid multipart bucket incarnation metadata"));
@@ -379,12 +379,12 @@ fn multipart_bucket_incarnation_id(metadata: &HashMap<String, String>) -> Result
    Ok(Some(incarnation))
 }

-fn multipart_bucket_incarnation_matches(metadata: &HashMap<String, String>, expected: Uuid) -> bool {
+fn multipart_bucket_incarnation_matches<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>, expected: Uuid) -> bool {
    matches!(multipart_bucket_incarnation_id(metadata), Ok(Some(actual)) if actual == expected)
 }

-fn validate_multipart_bucket_incarnation(
-    metadata: &HashMap<String, String>,
+fn validate_multipart_bucket_incarnation<S: std::hash::BuildHasher>(
+    metadata: &HashMap<String, String, S>,
    bucket: &str,
    object: &str,
    upload_id: &str,
@@ -1335,7 +1335,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                mod_time: Some(OffsetDateTime::now_utc()),
                actual_size,
                index: index_op,
-                checksums: if checksums.is_empty() { None } else { Some(checksums) },
+                checksums: if checksums.is_empty() { None } else { Some(checksums.iter().map(|(k, v)| (k.clone(), v.clone())).collect()) },
                ..Default::default()
            };

@@ -1523,7 +1523,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
            max_parts,
            part_number_marker,
            user_defined: {
-                let mut metadata = fi.metadata.clone();
+                let mut metadata: HashMap<String, String> = fi.metadata.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
                strip_internal_multipart_metadata(&mut metadata);
                metadata
            },
@@ -1782,7 +1782,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
        let mod_time = opts.mod_time.unwrap_or_else(OffsetDateTime::now_utc);

        for f in parts_metadatas.iter_mut() {
-            f.metadata = user_defined.clone();
+            f.metadata = user_defined.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
            f.mod_time = Some(mod_time);
            f.fresh = true;
        }
@@ -1871,7 +1871,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
            upload_id: upload_id.to_owned(),
            user_defined: {
                strip_internal_multipart_metadata(&mut fi.metadata);
-                fi.metadata.clone()
+                fi.metadata.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
            },
            ..Default::default()
        })
@@ -3693,18 +3693,18 @@ mod tests {
        }
    }

-    async fn object_transaction_epochs(
-        disks: &[DiskStore],
-        bucket: &str,
-        object: &str,
-        write_quorum: usize,
-    ) -> Vec<Option<Uuid>> {
+    async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec<Option<Uuid>> {
        let mut epochs = Vec::with_capacity(disks.len());
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
        for (disk_index, disk) in disks.iter().enumerate() {
-            let file_info = match disk.read_version("", bucket, object, "", &ReadOptions::default()).await {
-                Ok(file_info) => file_info,
-                Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => continue,
-                Err(err) => panic!("disk {disk_index} should read object metadata: {err}"),
+            let file_info = loop {
+                match disk.read_version("", bucket, object, "", &ReadOptions::default()).await {
+                    Ok(file_info) => break file_info,
+                    Err(DiskError::FileNotFound) if tokio::time::Instant::now() < deadline => {
+                        tokio::time::sleep(Duration::from_millis(25)).await;
+                    }
+                    Err(err) => panic!("disk {disk_index} should persist object metadata: {err}"),
+                }
            };
            epochs.push(
                file_info
@@ -3712,11 +3712,6 @@ mod tests {
                    .unwrap_or_else(|err| panic!("disk {disk_index} transaction epoch should decode: {err}")),
            );
        }
-        assert!(
-            epochs.len() >= write_quorum,
-            "object metadata should persist on write quorum: found {}, need {write_quorum}",
-            epochs.len()
-        );
        epochs
    }

@@ -3740,7 +3735,6 @@ mod tests {
            [
                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
-                (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("false")),
            ],
            async {
                set_disks
@@ -3774,26 +3768,34 @@ mod tests {
        let (upload_id, parts) =
            stage_upload_with_create_opts(&set_disks, bucket, object, b"multipart fenced epoch", &ObjectOptions::default()).await;

-        temp_env::async_with_vars(
+        let epochs = temp_env::async_with_vars(
            [
                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+                (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true")),
            ],
            async {
+                let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
                set_disks
                    .clone()
                    .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
                    .await
                    .expect("fenced multipart completion should commit with a live proof");
+
+                tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
+                    .await
+                    .expect("multipart completion should leave one rename tail in flight after quorum ACK");
+                let disks = disk_stores.clone();
+                let mut epochs = tokio::spawn(async move { object_transaction_epochs(&disks, bucket, object).await });
+                assert!(
+                    tokio::time::timeout(Duration::from_millis(100), &mut epochs).await.is_err(),
+                    "epoch read-back should wait for the lagging rename tail"
+                );
+                rename_barrier.release();
+                epochs.await.expect("epoch read-back should finish after the rename tail")
            },
        )
        .await;
-
-        disk_stores[0]
-            .delete_paths(bucket, &[format!("{object}/{STORAGE_FORMAT_FILE}")])
-            .await
-            .expect("one lagging disk should be simulated");
-        let epochs = object_transaction_epochs(&disk_stores, bucket, object, set_disks.default_write_quorum()).await;
        let first = epochs[0].expect("fenced multipart completion should persist an epoch");
        assert!(!first.is_nil());
        assert!(epochs.into_iter().all(|epoch| epoch == Some(first)));
@@ -3827,7 +3829,7 @@ mod tests {
                    )
                    .await
                    .expect("initial fenced PUT should commit");
-                let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object, set_disks.default_write_quorum())
+                let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object)
                    .await
                    .into_iter()
                    .next()
@@ -3869,7 +3871,7 @@ mod tests {
                    )
                    .await
                    .expect("concurrent fenced PUT should advance the epoch");
-                let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object, set_disks.default_write_quorum())
+                let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object)
                    .await
                    .into_iter()
                    .next()
@@ -3884,8 +3886,7 @@ mod tests {
                    .expect_err("stale epoch multipart completion must be rejected");
                assert_eq!(err, StorageError::PreconditionFailed);

-                let final_epochs =
-                    object_transaction_epochs(&disk_stores, bucket, object, set_disks.default_write_quorum()).await;
+                let final_epochs = object_transaction_epochs(&disk_stores, bucket, object).await;
                assert!(final_epochs.into_iter().all(|epoch| epoch == Some(winning_epoch)));
                let mut reader = set_disks
                    .get_object_reader(
@@ -19,6 +19,7 @@
 //! bounds are unchanged, and the impls reach shared primitives through the
 //! SetDisks core (io_primitives) via inherent calls.

+use ahash::AHashMap;
 #[cfg(test)]
 use super::super::MetadataCacheInvalidationProbe;
 use super::super::{
@@ -1107,13 +1108,13 @@ fn is_restore_control_metadata(key: &str) -> bool {
            .is_some_and(|remainder| remainder.is_empty())
 }

-fn restore_metadata_update_preserves_protected_metadata(
-    existing: &HashMap<String, String>,
-    replacement: &HashMap<String, String>,
+fn restore_metadata_update_preserves_protected_metadata<S1: std::hash::BuildHasher, S2: std::hash::BuildHasher>(
+    existing: &HashMap<String, String, S1>,
+    replacement: &HashMap<String, String, S2>,
 ) -> bool {
-    let mut existing = existing.clone();
+    let mut existing: HashMap<String, String> = existing.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
    clean_metadata(&mut existing);
-    let mut replacement = replacement.clone();
+    let mut replacement: HashMap<String, String> = replacement.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
    clean_metadata(&mut replacement);
    let existing_count = existing.keys().filter(|key| !is_restore_control_metadata(key)).count();
    let replacement_count = replacement.keys().filter(|key| !is_restore_control_metadata(key)).count();
@@ -2211,9 +2212,9 @@ pub(in crate::set_disk) fn stored_replication_category_metadata(existing: &Objec
 ///
 /// Returns whether `inbound` was modified. Callers must hold the object write
 /// lock so the stored values compared here are the ones being replaced.
-pub(in crate::set_disk) fn merge_replication_metadata_lww(
-    inbound: &mut HashMap<String, String>,
-    existing: &HashMap<String, String>,
+pub(in crate::set_disk) fn merge_replication_metadata_lww<S1: std::hash::BuildHasher, S2: std::hash::BuildHasher>(
+    inbound: &mut HashMap<String, String, S1>,
+    existing: &HashMap<String, String, S2>,
    opts: &ObjectOptions,
 ) -> bool {
    use rustfs_utils::http::headers::{
@@ -2844,7 +2845,7 @@ impl SetDisks {
                )));
            }

-            fi.metadata = user_defined;
+            fi.metadata = user_defined.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
            fi.mod_time = mod_time;
            fi.size = w_size as i64;
            fi.versioned = opts.versioned || opts.version_suspended;
@@ -6051,7 +6052,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
        } else {
            None
        };
-        let mut replacement_metadata = (*src_info.user_defined).clone();
+        let mut replacement_metadata: AHashMap<String, String> = (*src_info.user_defined).iter().map(|(k, v)| (k.clone(), v.clone())).collect();
        if let Some(part_checksums) = preserved_part_checksums {
            rustfs_utils::http::insert_str(&mut replacement_metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, part_checksums);
        }
@@ -7493,7 +7494,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
            X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str(),
            X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str(),
        ] {
-            if let Some(value) = fi.metadata.lookup(header).filter(|value| !value.is_empty()) {
+            if let Some(value) = fi.metadata.get(header).filter(|value| !value.is_empty()) {
                transition_meta.insert(header.to_ascii_lowercase(), value.to_string());
            }
        }
@@ -26,18 +26,18 @@ static STRICT_BUCKET_NAME_REGEX: LazyLock<Regex> =
 static NON_STRICT_BUCKET_NAME_REGEX: LazyLock<Regex> =
    LazyLock::new(|| Regex::new(r"^[A-Za-z0-9][A-Za-z0-9\.\-_:]{1,61}[A-Za-z0-9]$").expect("valid non-strict bucket name regex"));

-pub fn clean_metadata(metadata: &mut HashMap<String, String>) {
+pub fn clean_metadata<S: std::hash::BuildHasher>(metadata: &mut HashMap<String, String, S>) {
    remove_standard_storage_class(metadata);
    clean_metadata_keys(metadata, &["md5Sum", "etag", "expires", AMZ_OBJECT_TAGGING, "last-modified"]);
 }

-pub fn remove_standard_storage_class(metadata: &mut HashMap<String, String>) {
+pub fn remove_standard_storage_class<S: std::hash::BuildHasher>(metadata: &mut HashMap<String, String, S>) {
    if metadata.get(AMZ_STORAGE_CLASS) == Some(&STANDARD.to_string()) {
        metadata.remove(AMZ_STORAGE_CLASS);
    }
 }

-pub fn clean_metadata_keys(metadata: &mut HashMap<String, String>, key_names: &[&str]) {
+pub fn clean_metadata_keys<S: std::hash::BuildHasher>(metadata: &mut HashMap<String, String, S>, key_names: &[&str]) {
    for key in key_names {
        metadata.remove(key.to_owned());
    }
@@ -51,6 +51,9 @@ s3s = { workspace = true, features = ["minio"] }
 regex.workspace = true
 arc-swap.workspace = true

+# High-performance hashing
+ahash = { workspace = true, features = ["serde"] }
+
 [dev-dependencies]
 criterion = { workspace = true, features = ["html_reports"] }
 tempfile = { workspace = true }
@@ -22,6 +22,7 @@ use rustfs_utils::http::{
    contains_key_str, get_consistent_str, get_str, has_internal_suffix, insert_str, is_encryption_metadata_key,
    starts_with_ignore_ascii_case,
 };
+use ahash::AHashMap;
 use s3s::dto::{RestoreStatus, Timestamp};
 use s3s::header::X_AMZ_RESTORE;
 use serde::de::{self, MapAccess, SeqAccess, Visitor, value::MapAccessDeserializer};
@@ -67,7 +68,7 @@ pub struct ObjectPartInfo {
    // Index holds the index of the part in the erasure coding
    pub index: Option<Bytes>,
    // Checksums holds checksums of the part
-    pub checksums: Option<HashMap<String, String>>,
+    pub checksums: Option<AHashMap<String, String>>,
    pub error: Option<String>,
 }

@@ -268,7 +269,7 @@ pub struct FileInfo {
    pub mode: Option<u32>,
    // WrittenByVersion is the unix time stamp of the version that created this version of the object
    pub written_by_version: Option<u64>,
-    pub metadata: HashMap<String, String>,
+    pub metadata: AHashMap<String, String>,
    pub parts: Vec<ObjectPartInfo>,
    pub erasure: ErasureInfo,
    // MarkDeleted marks this version as deleted
@@ -301,7 +302,7 @@ fn is_sensitive_metadata_key(key: &str) -> bool {
            .any(|prefix| starts_with_ignore_ascii_case(key, prefix))
 }

-struct RedactedMetadata<'a>(&'a HashMap<String, String>);
+struct RedactedMetadata<'a>(&'a AHashMap<String, String>);

 impl std::fmt::Debug for RedactedMetadata<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -425,7 +426,7 @@ struct FileInfoMapDef {
    size: i64,
    mode: Option<u32>,
    written_by_version: Option<u64>,
-    metadata: HashMap<String, String>,
+    metadata: AHashMap<String, String>,
    parts: Vec<ObjectPartInfo>,
    erasure: ErasureInfo,
    mark_deleted: bool,
@@ -1079,7 +1080,7 @@ impl FileInfo {
        mod_time: Option<OffsetDateTime>,
        actual_size: i64,
        index: Option<Bytes>,
-        checksums: Option<HashMap<String, String>>,
+        checksums: Option<AHashMap<String, String>>,
    ) {
        let part = ObjectPartInfo {
            etag,
@@ -1457,7 +1458,7 @@ pub fn parse_restore_obj_status(restore_hdr: &str) -> Result<RestoreStatus> {
    Err(Error::other(ERR_RESTORE_HDR_MALFORMED))
 }

-pub fn is_restored_object_on_disk(meta: &HashMap<String, String>) -> bool {
+pub fn is_restored_object_on_disk<S: std::hash::BuildHasher>(meta: &HashMap<String, String, S>) -> bool {
    if let Some(restore_hdr) = meta.get(X_AMZ_RESTORE.as_str())
        && let Ok(restore_status) = parse_restore_obj_status(restore_hdr)
    {
@@ -2135,7 +2136,7 @@ mod tests {
            -1_000_000i64..=1_000_000i64,
            optional_timestamp_strategy(),
            proptest::option::of(bytes_strategy(16)),
-            proptest::option::of(hash_map(small_string_strategy(), small_string_strategy(), 0..=3)),
+            proptest::option::of(hash_map(small_string_strategy(), small_string_strategy(), 0..=3).prop_map(|m| m.into_iter().collect::<AHashMap<String, String>>())),
            proptest::option::of(small_string_strategy()),
        )
            .prop_map(|(etag, number, size, actual_size, mod_time, index, checksums, error)| ObjectPartInfo {
@@ -2170,7 +2171,7 @@ mod tests {
                -1_000_000i64..=1_000_000i64,
                proptest::option::of(any::<u32>()),
                proptest::option::of(any::<u64>()),
-                hash_map(small_string_strategy(), small_string_strategy(), 0..=4),
+                hash_map(small_string_strategy(), small_string_strategy(), 0..=4).prop_map(|m| m.into_iter().collect::<AHashMap<String, String>>()),
                vec(object_part_info_strategy(), 0..=3),
                erasure_info_strategy(),
                any::<bool>(),
@@ -174,10 +174,10 @@ fn valid_target_delete_marker_version(arn: &str, version_id: &str) -> bool {
 /// included in the quorum hash, so such a divergence does surface — but as a
 /// quorum failure on an otherwise healthy object, which is not a state worth
 /// reaching. Merge the RPC metadata carrier instead, and only ever insert.
-fn persist_target_delete_marker_versions(
+fn persist_target_delete_marker_versions<S: std::hash::BuildHasher>(
    meta_sys: &mut HashMap<String, Vec<u8>>,
    versions: &HashMap<String, String>,
-    transport_metadata: &HashMap<String, String>,
+    transport_metadata: &HashMap<String, String, S>,
 ) {
    let mut bounded = BTreeMap::new();
    // A corrupt carrier means the dual internal prefixes disagreed. Do not merge
@@ -181,7 +181,7 @@ mod tests {
            data_dir: Some(data_dir),
            size: 64 * 1024,
            mod_time: Some(OffsetDateTime::now_utc()),
-            metadata,
+            metadata: metadata.into_iter().collect(),
            erasure: ErasureInfo {
                algorithm: ErasureAlgo::ReedSolomon.to_string(),
                data_blocks: 4,
@@ -26,7 +26,7 @@ use super::msgp_decode::{
    PrependByteReader, prealloc_hint, read_exact_vec, read_nil_or_array_len, read_nil_or_map_len, skip_msgp_value,
 };
 use super::*;
-use crate::{ChecksumInfo, TransitionVersionState};
+use crate::{AHashMap, ChecksumInfo, TransitionVersionState};
 use rustfs_utils::HashAlgorithm;
 use rustfs_utils::http::{
    RUSTFS_INTERNAL_PREFIX, SUFFIX_CRC, SUFFIX_FREE_VERSION, SUFFIX_INLINE_DATA, SUFFIX_PART_CHECKSUMS, SUFFIX_PURGESTATUS,
@@ -377,7 +377,7 @@ impl<'a> DerivedInternalMetadata<'a> {
    }
 }

-struct UniquePartChecksums(HashMap<String, String>);
+struct UniquePartChecksums(AHashMap<String, String>);

 impl<'de> serde::Deserialize<'de> for UniquePartChecksums {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
@@ -397,7 +397,7 @@ impl<'de> serde::Deserialize<'de> for UniquePartChecksums {
            where
                A: serde::de::SeqAccess<'de>,
            {
-                let mut checksums = HashMap::with_capacity(seq.size_hint().unwrap_or_default());
+                let mut checksums = AHashMap::with_capacity(seq.size_hint().unwrap_or_default());
                while let Some((key, value)) = seq.next_element::<(String, String)>()? {
                    if checksums.insert(key, value).is_some() {
                        return Err(serde::de::Error::custom("duplicate part checksum name"));
@@ -1477,7 +1477,7 @@ pub struct MetaObjectV1 {
    #[serde(rename = "Erasure")]
    pub erasure: MetaObjectV1Erasure,
    #[serde(rename = "Meta")]
-    pub meta: HashMap<String, String>,
+    pub meta: AHashMap<String, String>,
    #[serde(rename = "Parts")]
    pub parts: Vec<MetaObjectV1Part>,
    #[serde(rename = "VersionID")]
@@ -1543,7 +1543,7 @@ pub struct MetaObjectV1Part {
    #[serde(rename = "i")]
    pub index: Option<Bytes>,
    #[serde(rename = "crc")]
-    pub checksums: Option<HashMap<String, String>>,
+    pub checksums: Option<AHashMap<String, String>>,
    #[serde(rename = "err")]
    pub error: Option<String>,
 }
@@ -1887,7 +1887,7 @@ impl MetaObjectV1Part {
                "i" => self.index = Some(Bytes::from(read_msgp_bin(rd)?)),
                "crc" => {
                    let len = rmp::decode::read_map_len(rd)? as usize;
-                    let mut checksums = HashMap::with_capacity(prealloc_hint(len));
+                    let mut checksums = AHashMap::with_capacity(prealloc_hint(len));
                    for _ in 0..len {
                        checksums.insert(read_msgp_string(rd)?, read_msgp_string(rd)?);
                    }
@@ -2570,7 +2570,7 @@ impl MetaObject {
            Vec::new()
        };

-        let mut metadata = HashMap::with_capacity(self.meta_user.len() + self.meta_sys.len());
+        let mut metadata = AHashMap::with_capacity(self.meta_user.len() + self.meta_sys.len());
        for (k, v) in &self.meta_user {
            if k == AMZ_META_UNENCRYPTED_CONTENT_LENGTH || k == AMZ_META_UNENCRYPTED_CONTENT_MD5 {
                continue;
@@ -2861,7 +2861,7 @@ impl From<FileInfo> for MetaObject {
    }
 }

-fn get_internal_replication_state(metadata: &HashMap<String, String>) -> Option<ReplicationState> {
+fn get_internal_replication_state<S: std::hash::BuildHasher>(metadata: &HashMap<String, String, S>) -> Option<ReplicationState> {
    let mut rs = ReplicationState::default();
    let mut has = false;

@@ -2942,7 +2942,7 @@ impl MetaDeleteMarker {
    }

    pub fn into_fileinfo(&self, volume: &str, path: &str, _all_parts: bool) -> Result<FileInfo> {
-        let metadata = self
+        let metadata: AHashMap<String, String> = self
            .meta_sys
            .clone()
            .into_iter()
@@ -5500,10 +5500,10 @@ mod tests {
    /// entirely, silently dropping the whole legacy body on re-marshal.
    #[test]
    fn legacy_version_body_round_trips_through_encode() {
-        let mut meta = HashMap::new();
+        let mut meta = AHashMap::new();
        meta.insert("content-type".to_string(), "application/octet-stream".to_string());

-        let mut crc = HashMap::new();
+        let mut crc = AHashMap::new();
        crc.insert("crc32c".to_string(), "deadbeef".to_string());

        let legacy = MetaObjectV1 {
@@ -22,6 +22,12 @@ mod replication;

 pub mod test_data;

+/// High-performance HashMap type alias using ahash instead of SipHash.
+pub type AHashMap<K, V> = ahash::AHashMap<K, V>;
+
+/// High-performance HashSet type alias using ahash.
+pub type AHashSet<K> = ahash::AHashSet<K>;
+
 pub use error::*;
 pub use fileinfo::*;
 pub use filemeta::*;
@@ -14,8 +14,8 @@

 use crate::filemeta::msgp_decode::MAX_MSGP_ELEMENT_SIZE;
 use crate::{
-    Error, FileInfo, FileInfoOpts, FileInfoVersions, FileMeta, FileMetaShallowVersion, Result, VersionType, get_file_info,
-    merge_file_meta_versions, merge_file_meta_versions_with_write_quorum,
+    AHashMap, Error, FileInfo, FileInfoOpts, FileInfoVersions, FileMeta, FileMetaShallowVersion, Result, VersionType,
+    get_file_info, merge_file_meta_versions, merge_file_meta_versions_with_write_quorum,
 };
 use arc_swap::ArcSwapOption;
 use rmp::Marker;
@@ -1676,7 +1676,7 @@ mod tests {
    }

    fn metacache_entry_with_mod_time(mod_time: OffsetDateTime, etag: &str) -> MetaCacheEntry {
-        let mut metadata = HashMap::new();
+        let mut metadata = AHashMap::new();
        metadata.insert("etag".to_string(), etag.to_string());

        let mut meta = FileMeta::new();
@@ -1705,7 +1705,7 @@ mod tests {
        data_blocks: usize,
        parity_blocks: usize,
    ) -> MetaCacheEntry {
-        let mut metadata = HashMap::new();
+        let mut metadata = AHashMap::new();
        metadata.insert("etag".to_string(), etag.to_string());

        let mut fi = FileInfo::new("object", data_blocks, parity_blocks);
@@ -1730,7 +1730,7 @@ mod tests {
    fn metacache_entry_with_erasure_versions(versions: &[(OffsetDateTime, &str, usize, usize)]) -> MetaCacheEntry {
        let mut meta = FileMeta::new();
        for (idx, (mod_time, etag, data_blocks, parity_blocks)) in versions.iter().enumerate() {
-            let mut metadata = HashMap::new();
+            let mut metadata = AHashMap::new();
            metadata.insert("etag".to_string(), (*etag).to_string());

            let mut fi = FileInfo::new("object", *data_blocks, *parity_blocks);
@@ -1776,7 +1776,7 @@ mod tests {
    /// Build an entry holding a single object version with an explicit version id
    /// and mod_time, so a set of these can model DISJOINT per-disk version sets.
    fn metacache_entry_single_version(version_u128: u128, mod_time: OffsetDateTime, etag: &str) -> MetaCacheEntry {
-        let mut metadata = HashMap::new();
+        let mut metadata = AHashMap::new();
        metadata.insert("etag".to_string(), etag.to_string());

        let mut fi = FileInfo::new("object", 4, 2);
@@ -74,7 +74,7 @@ hotpath-cpu = [
 # Tokio runtime-level telemetry. Requires a `--cfg tokio_unstable` build; the
 # build script fails the compile when that flag is missing. Off by default so
 # ordinary builds neither pay for nor depend on Tokio's unstable API.
-dial9 = ["dep:dial9-tokio-telemetry"]
+dial9 = ["dep:dial9-tokio-telemetry", "dial9-tokio-telemetry/process-resource"]
 #
 # NOTE: there is deliberately no `dial9-taskdump` feature. dial9 only captures a
 # task dump for futures it wrapped itself, i.e. those spawned via
@@ -76,7 +76,7 @@ pub struct Dial9Config {
    /// Directory where trace files are written
    pub output_dir: String,

-    /// Prefix for trace file names
+    /// Trace family name under the output directory
    pub file_prefix: String,

    /// Maximum size of each trace file in bytes
@@ -158,7 +158,7 @@ impl Dial9Config {
        }
    }

-    /// Get the base path for trace files.
+    /// Get the trace family directory for rotating trace segments.
    pub fn base_path(&self) -> PathBuf {
        PathBuf::from(&self.output_dir).join(&self.file_prefix)
    }
@@ -23,15 +23,22 @@ use super::config::Dial9Config;
 use super::state::{dial9_runtime_state, measure_disk_usage_bytes};
 use super::{EVENT_DIAL9_STATE, LOG_COMPONENT_OBS, LOG_SUBSYSTEM_DIAL9};
 use crate::TelemetryError;
-use dial9_tokio_telemetry::telemetry::{ProcessResourceUsageConfig, RotatingWriter, TracedRuntime};
+use dial9_tokio_telemetry::telemetry::{
+    Dial9Handle, Dial9HandleTokioExt, DiskBuffer, ProcessResourceUsageConfig, RecorderPerfExt, TokioAttachOptions, recorder,
+};
 use std::time::Duration;
 use tracing::{info, warn};

-pub use dial9_tokio_telemetry::telemetry::TelemetryGuard;
+pub type TelemetryGuard = Dial9Handle;
+
+type ShutdownRecorder = Box<dyn FnOnce() + Send + 'static>;

 /// How often the background refresher restates trace-file disk usage.
 const DISK_USAGE_REFRESH_INTERVAL: Duration = Duration::from_secs(60);

+/// Maximum time spent flushing the recorder during graceful shutdown.
+const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
+
 /// Name recorded in segment metadata so the trace viewer can label workers.
 const RUNTIME_NAME: &str = "rustfs-worker";

@@ -43,13 +50,14 @@ const RUNTIME_NAME: &str = "rustfs-worker";
 /// are lost.
 pub struct Dial9SessionGuard {
    guard: TelemetryGuard,
+    shutdown: Option<ShutdownRecorder>,
    config: Dial9Config,
 }

 impl Dial9SessionGuard {
    /// Whether the underlying telemetry session is recording.
    pub fn is_active(&self) -> bool {
-        self.guard.is_enabled()
+        self.guard.is_enabled() && self.guard.is_connected() && !self.guard.is_stopped()
    }
 }

@@ -72,8 +80,10 @@ impl Drop for Dial9SessionGuard {
            state = "flushed",
            "dial9 state changed"
        );
-        // `TelemetryGuard`'s own `Drop` flushes buffered events and seals the
-        // active segment; it runs immediately after this body.
+
+        if let Some(shutdown) = self.shutdown.take() {
+            shutdown();
+        }
    }
 }

@@ -97,53 +107,58 @@ pub fn build_traced_runtime(
        TelemetryError::Io(format!("Failed to create dial9 output directory '{}': {e}", config.output_dir))
    })?;

-    let writer = RotatingWriter::new(config.base_path(), config.max_file_size, config.total_disk_budget()).map_err(|e| {
-        dial9_runtime_state().record_runtime_error(&config);
-        TelemetryError::Io(format!("Failed to create dial9 RotatingWriter: {e}"))
-    })?;
+    let writer = DiskBuffer::builder()
+        .base_path(config.base_path())
+        .max_file_size(config.max_file_size)
+        .max_total_size(config.total_disk_budget())
+        .build()
+        .map_err(|e| {
+            dial9_runtime_state().record_runtime_error(&config);
+            TelemetryError::Io(format!("Failed to create dial9 DiskBuffer: {e}"))
+        })?;

-    // `with_trace_path` transitions the builder into the state that spawns the
-    // background worker, which drives the segment pipeline.
-    let traced = TracedRuntime::builder()
-        .with_trace_path(&config.output_dir)
-        .with_task_tracking(true)
-        .with_runtime_name(RUNTIME_NAME)
-        .with_process_resource_usage(ProcessResourceUsageConfig::default());
+    let recorder = recorder(writer)
+        .with_process_resource_usage(ProcessResourceUsageConfig::default())
+        .build();
+    let guard = recorder.handle().clone();
+    let shutdown: ShutdownRecorder = Box::new(move || recorder.graceful_shutdown(SHUTDOWN_TIMEOUT));

-    // `build_and_start` rather than `build`: `build` returns a live guard that
-    // never records, writing segments that contain only a header.
-    //
-    // No `with_task_dumps` here. dial9 captures a task dump only for futures it
+    let attached = guard
+        .attach_tokio_runtime(
+            builder,
+            TokioAttachOptions::builder()
+                .runtime_name(RUNTIME_NAME)
+                .task_tracking_enabled(true)
+                .build(),
+        )
+        .map(|runtime| (runtime, guard, shutdown));
+
+    // No task dumps here. dial9 captures a task dump only for futures it
    // wrapped itself, i.e. those spawned via `dial9_tokio_telemetry::spawn`;
    // `tokio::spawn` gets no wrapper. RustFS spawns with `tokio::spawn`
-    // throughout, so calling `with_task_dumps` records nothing. Measured on an
+    // throughout, so enabling task dumps records nothing. Measured on an
    // identical workload: 0 dumps via `tokio::spawn`, 14709 via `dial9::spawn`.
    // See rustfs/backlog#1157 (D9-16) and dial9-rs/dial9#477.
    //
    // No `with_s3_uploader` here: dial9's `worker-s3` feature carries a
    // vulnerable TLS stack. See the note in `crates/obs/Cargo.toml`.
-    finish_traced_runtime(traced.build_and_start(builder, writer), config)
+    finish_traced_runtime(attached, config)
 }

 /// Publish the outcome of a traced-runtime build and start the background
 /// disk-usage refresher.
 fn finish_traced_runtime(
-    started: std::io::Result<(tokio::runtime::Runtime, TelemetryGuard)>,
+    started: std::io::Result<(tokio::runtime::Runtime, TelemetryGuard, ShutdownRecorder)>,
    config: Dial9Config,
 ) -> Result<(tokio::runtime::Runtime, Dial9SessionGuard), TelemetryError> {
-    let (runtime, guard) = started.map_err(|e| {
+    let (runtime, guard, shutdown) = started.map_err(|e| {
        dial9_runtime_state().record_runtime_error(&config);
-        TelemetryError::Io(format!("Failed to build dial9 TracedRuntime: {e}"))
+        TelemetryError::Io(format!("Failed to attach dial9 runtime telemetry: {e}"))
    })?;

-    // `is_enabled` distinguishes a live guard from the inert one a lenient
-    // config produces after a build failure. It does NOT mean recording has
-    // started — a guard from `build` (rather than `build_and_start`) reports
-    // `true` while writing segments that contain only a header. Recording is
-    // guaranteed by the `build_and_start` call above, not by this check.
    if !guard.is_enabled() {
        dial9_runtime_state().record_runtime_error(&config);
-        return Err(TelemetryError::Io("dial9 TracedRuntime built with telemetry disabled".to_string()));
+        return Err(TelemetryError::Io("dial9 runtime telemetry attached with recording disabled".to_string()));
    }

    dial9_runtime_state().record_runtime_started(&config);
@@ -160,7 +175,14 @@ fn finish_traced_runtime(
        "dial9 state changed"
    );

-    Ok((runtime, Dial9SessionGuard { guard, config }))
+    Ok((
+        runtime,
+        Dial9SessionGuard {
+            guard,
+            shutdown: Some(shutdown),
+            config,
+        },
+    ))
 }

 /// Periodically restate trace-file disk usage so the metrics collector can read
@@ -49,13 +49,13 @@
 //!
 //! # Known observability gap
 //!
-//! `dial9`'s `RotatingWriter` stops accepting writes (its internal `Finished`
-//! state) when the output directory disappears or a segment cannot be sealed,
-//! and it exposes no way to observe that from outside. `TelemetryGuard::is_enabled`
-//! reports how the session was *built*, not whether it is still writing. There
-//! is therefore no `writer_healthy` metric: it could only ever be hard-coded to
-//! `1`. Watch `rustfs_dial9_disk_usage_bytes` — a session that is recording but
-//! whose disk usage stops growing has most likely hit this state.
+//! `dial9`'s `DiskBuffer` stops accepting writes when the output directory
+//! disappears or a segment cannot be sealed, and it exposes no way to observe
+//! that from outside. `Dial9Handle::is_enabled` reports whether the recorder is
+//! connected and unpaused, not whether the disk writer is still making progress.
+//! There is therefore no `writer_healthy` metric: it could only ever be
+//! hard-coded to `1`. Watch `rustfs_dial9_disk_usage_bytes` — a session that is
+//! recording but whose disk usage stops growing has most likely hit this state.
 //! Reported upstream as dial9-rs/dial9#658.

 mod config;
@@ -27,6 +27,9 @@ use std::sync::OnceLock;
 use std::sync::RwLock;
 use std::sync::atomic::{AtomicU64, Ordering};

+/// Segment filename stem used by `dial9` rotating disk buffers.
+const DIAL9_SEGMENT_STEM: &str = "trace";
+
 /// Point-in-time view of dial9 runtime state.
 #[derive(Debug, Clone, Default)]
 pub(crate) struct Dial9RuntimeSnapshot {
@@ -66,8 +69,8 @@ impl Dial9RuntimeState {

    pub(super) fn record_config(&self, config: &Dial9Config) {
        *self.trace_dir.write().expect("dial9 trace_dir lock should not be poisoned") = Some(TraceLocation {
-            output_dir: PathBuf::from(&config.output_dir),
-            file_prefix: config.file_prefix.clone(),
+            output_dir: config.base_path(),
+            file_prefix: DIAL9_SEGMENT_STEM.to_string(),
        });
        if !config.enabled {
            self.active_sessions.store(0, Ordering::Relaxed);
@@ -168,11 +171,11 @@ mod tests {
    #[test]
    fn measure_disk_usage_sums_only_matching_prefix() {
        let dir = tempdir().expect("create temp dir");
-        std::fs::write(dir.path().join("rustfs-tokio.0.bin"), vec![0_u8; 128]).expect("write segment");
-        std::fs::write(dir.path().join("rustfs-tokio.1.bin"), vec![0_u8; 64]).expect("write segment");
+        std::fs::write(dir.path().join("trace.0.bin"), vec![0_u8; 128]).expect("write segment");
+        std::fs::write(dir.path().join("trace.1.bin"), vec![0_u8; 64]).expect("write segment");
        std::fs::write(dir.path().join("unrelated.log"), vec![0_u8; 4096]).expect("write unrelated");

-        assert_eq!(measure_disk_usage_bytes(dir.path(), "rustfs-tokio"), 192);
+        assert_eq!(measure_disk_usage_bytes(dir.path(), DIAL9_SEGMENT_STEM), 192);
    }

    #[test]
@@ -54,7 +54,6 @@ rustls-pki-types.workspace = true
 s3s = { workspace = true, features = ["minio"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, features = ["raw_value"] }
-sha1 = { workspace = true }
 sha2 = { workspace = true }
 thiserror.workspace = true
 time = { workspace = true, features = ["parsing", "formatting", "macros", "serde"] }
@@ -58,6 +58,9 @@ transform-stream = { workspace = true, optional = true }
 url = { workspace = true, optional = true }
 zstd = { workspace = true, optional = true }

+# High-performance hashing
+ahash = { workspace = true, optional = true }
+
 [dev-dependencies]
 criterion = { workspace = true, features = ["html_reports"] }
 tempfile = { workspace = true }
@@ -89,7 +89,7 @@ pub fn is_object_encryption_marker(key: &str) -> bool {
 }

 /// Reads the logical object size recorded by encryption metadata.
-pub fn get_object_encryption_original_size(metadata: &std::collections::HashMap<String, String>) -> std::io::Result<Option<i64>> {
+pub fn get_object_encryption_original_size<S: std::hash::BuildHasher>(metadata: &std::collections::HashMap<String, String, S>) -> std::io::Result<Option<i64>> {
    let actual_size = super::get_str(metadata, super::SUFFIX_ACTUAL_SIZE);
    let size = get_case_insensitive(metadata, RUSTFS_ENCRYPTION_ORIGINAL_SIZE)
        .or_else(|| get_case_insensitive(metadata, SSEC_ORIGINAL_SIZE))
@@ -103,7 +103,7 @@ pub fn get_object_encryption_original_size(metadata: &std::collections::HashMap<
        .map_err(|error| std::io::Error::other(format!("Failed to parse encryption original size: {error}")))
 }

-fn get_case_insensitive<'a>(metadata: &'a std::collections::HashMap<String, String>, key: &str) -> Option<&'a str> {
+fn get_case_insensitive<'a, S: std::hash::BuildHasher>(metadata: &'a std::collections::HashMap<String, String, S>, key: &str) -> Option<&'a str> {
    metadata.get(key).map(String::as_str).or_else(|| {
        metadata
            .iter()
@@ -182,13 +182,13 @@ pub fn internal_key_rustfs(suffix: &str) -> String {

 // === String type (FileInfo.metadata, user_defined) ===

-pub fn insert_str(map: &mut HashMap<String, String>, suffix: &str, value: String) {
+pub fn insert_str<S: std::hash::BuildHasher>(map: &mut HashMap<String, String, S>, suffix: &str, value: String) {
    let (k1, k2) = both_keys(suffix);
    map.insert(k1, value.clone());
    map.insert(k2, value);
 }

-pub fn get_str(map: &HashMap<String, String>, suffix: &str) -> Option<String> {
+pub fn get_str<S: std::hash::BuildHasher>(map: &HashMap<String, String, S>, suffix: &str) -> Option<String> {
    if let Some(v) = with_internal_key(RUSTFS_INTERNAL_PREFIX, suffix, |k1| map.get(k1).cloned()) {
        return Some(v);
    }
@@ -202,7 +202,7 @@ pub fn get_str(map: &HashMap<String, String>, suffix: &str) -> Option<String> {
        .map(|(_, value)| value.clone())
 }

-fn get_consistent_value<'a, V: AsRef<[u8]>>(map: &'a HashMap<String, V>, suffix: &str) -> Option<&'a V> {
+fn get_consistent_value<'a, V: AsRef<[u8]>, S: std::hash::BuildHasher>(map: &'a HashMap<String, V, S>, suffix: &str) -> Option<&'a V> {
    let (rustfs_key, minio_key) = both_keys(suffix);
    let mut value = None;
    for (key, candidate) in map {
@@ -220,11 +220,11 @@ fn get_consistent_value<'a, V: AsRef<[u8]>>(map: &'a HashMap<String, V>, suffix:
 /// Returns a non-empty value when every compatibility key present for `suffix` agrees.
 /// A single RustFS or MinIO key is accepted for backward compatibility; conflicting or empty
 /// values return `None` so callers at destructive boundaries can fail closed.
-pub fn get_consistent_str<'a>(map: &'a HashMap<String, String>, suffix: &str) -> Option<&'a str> {
+pub fn get_consistent_str<'a, S: std::hash::BuildHasher>(map: &'a HashMap<String, String, S>, suffix: &str) -> Option<&'a str> {
    get_consistent_value(map, suffix).map(String::as_str)
 }

-pub fn contains_key_str(map: &HashMap<String, String>, suffix: &str) -> bool {
+pub fn contains_key_str<S: std::hash::BuildHasher>(map: &HashMap<String, String, S>, suffix: &str) -> bool {
    if with_internal_key(RUSTFS_INTERNAL_PREFIX, suffix, |k1| map.contains_key(k1)) {
        return true;
    }
@@ -236,7 +236,7 @@ pub fn contains_key_str(map: &HashMap<String, String>, suffix: &str) -> bool {
        .any(|key| key.eq_ignore_ascii_case(&k1) || key.eq_ignore_ascii_case(&k2))
 }

-pub fn remove_str(map: &mut HashMap<String, String>, suffix: &str) {
+pub fn remove_str<S: std::hash::BuildHasher>(map: &mut HashMap<String, String, S>, suffix: &str) {
    with_internal_key(RUSTFS_INTERNAL_PREFIX, suffix, |k1| map.remove(k1));
    with_internal_key(MINIO_INTERNAL_PREFIX, suffix, |k2| map.remove(k2));
    let (k1, k2) = both_keys(suffix);
@@ -285,7 +285,7 @@ pub fn strip_internal_prefix_preserving_case(key: &str) -> Option<&str> {

 /// Reads the bounded per-target delete-marker version map in one metadata scan.
 /// The boolean is set when matching metadata is malformed or compatibility keys disagree.
-pub fn target_delete_marker_versions(map: &HashMap<String, String>) -> (HashMap<String, String>, bool) {
+pub fn target_delete_marker_versions<S: std::hash::BuildHasher>(map: &HashMap<String, String, S>) -> (HashMap<String, String>, bool) {
    const MAX_ENTRIES: usize = 1_000;
    const MAX_ARN_LEN: usize = 1_024;
    const MAX_VERSION_ID_LEN: usize = 1_024;
@@ -0,0 +1,35 @@
+# Presigned PutObject size limit
+
+RustFS V1 supports an optional, RustFS-specific capability on a SigV4
+presigned `PutObject` URL:
+
+```text
+x-rustfs-max-content-length=<unsigned 64-bit integer>
+```
+
+The backend that creates the URL must add this query parameter to the request
+URI before calculating the SigV4 presign. It is part of the canonical query;
+adding, removing, or changing it after signing invalidates the signature. A
+browser can then upload with a plain `PUT` and does not need a custom size
+header.
+
+RustFS validates the capability after SigV4 authentication and enforces it on
+the decoded request body. A declared `Content-Length` above the limit is
+rejected before storage. If the body produces more bytes than the limit while
+streaming, RustFS returns `EntityTooLarge` and does not publish the object.
+
+The V1 contract is deliberately narrow:
+
+- The parameter is accepted only on a SigV4 presigned `PutObject` request.
+- Duplicate, case-variant, malformed, negative, or overflowing values return
+  `InvalidRequest`.
+- Requests without the parameter, including ordinary authenticated or
+  anonymous `PUT`, keep the existing behavior.
+- The parameter on `CopyObject`, multipart, `GET`, `HEAD`, `DELETE`, bucket, or
+  other operations returns `InvalidRequest`.
+- Unknown-length and SigV4 streaming-chunked uploads remain unsupported by the
+  existing PutObject admission contract and are not enabled by this feature.
+
+This capability is per request; it is not a cumulative multipart-upload cap.
+Multipart session limits are planned for V2 under a separate query/API
+contract.
@@ -78,6 +78,7 @@ use crate::app::object_usecase::{
 use crate::app::runtime_sources::{
    AppContext, current_app_context, current_object_data_cache_for_context, current_object_store_handle_for_context,
 };
+use crate::auth::{VerifiedPresignedRequest, reject_presigned_put_max_content_length_for_other_operation};
 use crate::capacity::record_capacity_write;
 use crate::error::ApiError;
 use crate::table_catalog;
@@ -397,6 +398,11 @@ impl DefaultMultipartUsecase {
        &self,
        req: S3Request<AbortMultipartUploadInput>,
    ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        record_s3_op(S3Operation::AbortMultipartUpload);
        let mut opts = ObjectOptions::default();
        apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
@@ -438,6 +444,11 @@ impl DefaultMultipartUsecase {
        &self,
        req: S3Request<CompleteMultipartUploadInput>,
    ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        let mut helper = OperationHelper::new(
            &req,
            EventName::ObjectCreatedCompleteMultipartUpload,
@@ -741,6 +752,11 @@ impl DefaultMultipartUsecase {
        &self,
        req: S3Request<CreateMultipartUploadInput>,
    ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        let helper =
            OperationHelper::new(&req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload)
                .suppress_event();
@@ -962,6 +978,11 @@ impl DefaultMultipartUsecase {
    #[instrument(level = "debug", skip(self, req))]
    #[hotpath::measure(impl_type = "MultipartUsecase")]
    pub async fn execute_upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        let mut opts = ObjectOptions::default();
        apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
        let input = req.input;
@@ -1229,6 +1250,11 @@ impl DefaultMultipartUsecase {
        &self,
        req: S3Request<ListMultipartUploadsInput>,
    ) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        let mut opts = ObjectOptions::default();
        apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
        let ListMultipartUploadsInput {
@@ -1276,6 +1302,11 @@ impl DefaultMultipartUsecase {
    }

    pub async fn execute_list_parts(&self, req: S3Request<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        let mut opts = ObjectOptions::default();
        apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
        let ListPartsInput {
@@ -1307,6 +1338,11 @@ impl DefaultMultipartUsecase {
        &self,
        req: S3Request<UploadPartCopyInput>,
    ) -> S3Result<S3Response<UploadPartCopyOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        // Captured before `req.input` is destructured below.
        let copy_principal = SseKmsPrincipal::from_request(&req);
        let source_bucket = match &req.input.copy_source {
@@ -16,6 +16,8 @@

 use super::*;

+use crate::auth::{VerifiedPresignedRequest, reject_presigned_put_max_content_length_for_other_operation};
+
 fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError {
    match err {
        rustfs_lock::LockError::QuorumNotReached { required, achieved } => StorageError::NamespaceLockQuorumUnavailable {
@@ -92,6 +94,11 @@ impl DefaultObjectUsecase {

    #[instrument(name = "execute_copy_object", level = "debug", skip(self, req))]
    async fn execute_copy_object_inner(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        if let Some(context) = &self.context {
            let _ = context.object_store();
        }
@@ -7242,6 +7242,12 @@ mod tests {
        let mut staged_targets = Vec::with_capacity(pool_disk_paths[target_pool].len());
        for (source_disk, target_disk) in pool_disk_paths[source_pool].iter().zip(&pool_disk_paths[target_pool]) {
            let source_dir = source_disk.join(&bucket).join(object);
+            // The same write-quorum minority gap tolerated above (#6701) can
+            // leave a lagging source-pool disk without the object; skip it and
+            // stage the replicas that exist — the reader tolerates the gap.
+            if !source_dir.join("xl.meta").is_file() {
+                continue;
+            }
            let target_dir = target_disk.join(&bucket).join(object);
            let staging_dir = temp_dir.path().join(format!("resume-relocate-{}", Uuid::new_v4()));
            std::fs::create_dir_all(&staging_dir).expect("create relocated target staging directory");
@@ -7258,15 +7264,18 @@ mod tests {
                }
            }
            std::fs::copy(source_dir.join("xl.meta"), staging_dir.join("xl.meta")).expect("stage relocated object metadata");
-            staged_targets.push((staging_dir, target_dir));
+            staged_targets.push((staging_dir, target_dir, source_dir.join("xl.meta")));
        }
+        assert!(
+            staged_targets.len() > pool_disk_paths[source_pool].len() / 2,
+            "a write-quorum majority of the source pool's disks must hold the object to stage the relocation"
+        );
        let (version_dirs, deleted) = delete_object_part_shards(&pool_disk_paths[source_pool], &bucket, object, &[2, 3]);
        assert!(version_dirs > 0, "the source pool must have at least one version data directory");
        assert_eq!(deleted, version_dirs * 2);

-        for ((staging_dir, target_dir), source_disk) in staged_targets.into_iter().zip(&pool_disk_paths[source_pool]) {
+        for (staging_dir, target_dir, source_meta) in staged_targets {
            std::fs::rename(staging_dir, target_dir).expect("publish relocated target object");
-            let source_meta = source_disk.join(&bucket).join(object).join("xl.meta");
            std::fs::remove_file(source_meta).expect("remove relocated source object metadata");
        }
        store
@@ -16,6 +16,9 @@

 use super::*;

+use crate::auth::{RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, parse_presigned_put_max_content_length};
+use crate::error::UploadLimitExceeded;
+
 const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024;

 const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES";
@@ -124,6 +127,58 @@ struct RequestBodyReadTimeout {
    timed_out: bool,
 }

+/// Enforces a maximum size on the decoded request entity while preserving the
+/// streaming behavior of the underlying S3 body.
+struct MaxContentLengthStream {
+    inner: StreamingBlob,
+    limit: u64,
+    received: u64,
+    exceeded: bool,
+}
+
+impl Stream for MaxContentLengthStream {
+    type Item = Result<Bytes, StdError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.as_mut().get_mut();
+        if this.exceeded {
+            return Poll::Ready(None);
+        }
+
+        match Pin::new(&mut this.inner).poll_next(cx) {
+            Poll::Ready(Some(Ok(chunk))) => {
+                let chunk_len = u64::try_from(chunk.len()).unwrap_or(u64::MAX);
+                let exceeds = this.received > this.limit || chunk_len > this.limit.saturating_sub(this.received);
+                if exceeds {
+                    this.exceeded = true;
+                    return Poll::Ready(Some(Err(Box::new(UploadLimitExceeded { limit: this.limit }))));
+                }
+
+                this.received = this.received.saturating_add(chunk_len);
+                Poll::Ready(Some(Ok(chunk)))
+            }
+            other => other,
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let remaining = usize::try_from(self.limit.saturating_sub(self.received)).unwrap_or(usize::MAX);
+        let (lower, upper) = self.inner.size_hint();
+        (lower.min(remaining), upper.map(|upper| upper.min(remaining)))
+    }
+}
+
+impl ByteStream for MaxContentLengthStream {
+    fn remaining_length(&self) -> RemainingLength {
+        let remaining = usize::try_from(self.limit.saturating_sub(self.received)).unwrap_or(usize::MAX);
+        let inner = self.inner.remaining_length();
+        inner
+            .exact()
+            .map(|exact| RemainingLength::new_exact(exact.min(remaining)))
+            .unwrap_or_else(RemainingLength::unknown)
+    }
+}
+
 impl Stream for RequestBodyReadTimeout {
    type Item = Result<Bytes, StdError>;

@@ -752,6 +807,11 @@ impl DefaultObjectUsecase {
        }

        let (event_name, quota_operation, request_method_name) = Self::put_object_execution_context(&req);
+        let max_content_length = parse_presigned_put_max_content_length(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
        if req.extensions.get::<PostObjectRequestMarker>().is_some() && is_post_object_sse_kms_requested(&req.input, &req.headers)
        {
            return Err(s3_error!(NotImplemented, "SSE-KMS is not supported for POST object uploads"));
@@ -769,6 +829,12 @@ impl DefaultObjectUsecase {
        // member) instead of writing the replica.
        let inbound_replication_put = replication_request_authorized(&req)
            && get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true");
+        if max_content_length.is_some() && is_put_object_extract_requested(&req.headers) {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is not supported for archive extraction"),
+            ));
+        }
        if is_put_object_extract_requested(&req.headers) && !inbound_replication_put {
            return Box::pin(self.execute_put_object_extract(req)).await;
        }
@@ -842,9 +908,25 @@ impl DefaultObjectUsecase {
            guard_put_object_body_read_timeout(body, &bucket, &key, &request_id, content_length, put_object_body_read_timeout())
        };

+        let body = match max_content_length {
+            Some(limit) => StreamingBlob::new(MaxContentLengthStream {
+                inner: body,
+                limit,
+                received: 0,
+                exceeded: false,
+            }),
+            None => body,
+        };
+
        // Resolve the authoritative decoded/plain object length (rejecting negative/unknown) before anything else consumes it.
        let mut size = resolve_put_object_authoritative_size(&req.headers, content_length)?;

+        if let Some(limit) = max_content_length
+            && u64::try_from(size).is_ok_and(|size| size > limit)
+        {
+            return Err(S3Error::new(S3ErrorCode::EntityTooLarge));
+        }
+
        // The app check preserves the existing S3 error contract; the storage
        // commit path reserves the exact net logical growth under its locks.
        let quota_check = self
@@ -1555,6 +1637,7 @@ pub(super) fn previous_current_size_from_backfill(backfill: Option<OldCurrentSiz
 #[cfg(test)]
 mod tests {
    use super::*;
+    use futures::StreamExt;
    use http::{HeaderMap, HeaderName, HeaderValue, Method};
    use s3s::dto::{DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRule};
    use std::pin::Pin;
@@ -1594,6 +1677,39 @@ mod tests {
        .expect("cancelled owner must abort and reap the stalled storage task");
    }

+    #[tokio::test]
+    async fn max_content_length_stream_rejects_the_first_chunk_over_limit() {
+        let inner = StreamingBlob::wrap(futures::stream::iter([
+            Ok::<Bytes, std::io::Error>(Bytes::from_static(b"1234")),
+            Ok::<Bytes, std::io::Error>(Bytes::from_static(b"56")),
+        ]));
+        let mut limited = MaxContentLengthStream {
+            inner,
+            limit: 5,
+            received: 0,
+            exceeded: false,
+        };
+
+        assert_eq!(limited.next().await.unwrap().unwrap(), Bytes::from_static(b"1234"));
+        let error = limited.next().await.unwrap().unwrap_err();
+        assert!(error.downcast_ref::<UploadLimitExceeded>().is_some());
+        assert!(limited.next().await.is_none());
+    }
+
+    #[tokio::test]
+    async fn max_content_length_stream_allows_exact_limit() {
+        let inner = StreamingBlob::from_bytes(Bytes::from_static(b"12345"));
+        let mut limited = MaxContentLengthStream {
+            inner,
+            limit: 5,
+            received: 0,
+            exceeded: false,
+        };
+
+        assert_eq!(limited.next().await.unwrap().unwrap(), Bytes::from_static(b"12345"));
+        assert!(limited.next().await.is_none());
+    }
+
    #[test]
    fn put_request_user_metadata_cannot_suppress_bucket_default_retention() {
        let mut metadata =
@@ -38,6 +38,7 @@ use subtle::ConstantTimeEq;
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
 use tracing::{debug, trace, warn};
+use url::form_urlencoded;

 const LOG_COMPONENT_AUTH: &str = "auth";
 const LOG_SUBSYSTEM_CREDENTIALS: &str = "credentials";
@@ -50,6 +51,15 @@ const EVENT_KEYSTONE_CREDENTIALS_VALIDATED: &str = "keystone_credentials_validat
 const EVENT_KEYSTONE_CONTEXT_MISSING: &str = "keystone_context_missing";
 const EVENT_SESSION_TOKEN_EXTRACTION: &str = "session_token_extraction";

+/// RustFS-specific query capability for a single presigned PutObject request.
+pub(crate) const RUSTFS_MAX_CONTENT_LENGTH_QUERY: &str = "x-rustfs-max-content-length";
+
+/// Inserted by the S3 access boundary after the upstream verifier accepts a
+/// request as SigV4 presigned. Downstream capability parsing must require this
+/// marker instead of treating query syntax as proof of authentication.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct VerifiedPresignedRequest;
+
 /// Performs constant-time string comparison to prevent timing attacks.
 ///
 /// This function should be used when comparing sensitive values like passwords,
@@ -913,9 +923,11 @@ pub(crate) fn is_request_presigned_signature_v4_with_query(header: &HeaderMap, q
    if let Some(credential) = header.get(AMZ_CREDENTIAL) {
        return !credential.to_str().unwrap_or("").is_empty();
    }
-    query
-        .and_then(|query| get_query_param(query, "x-amz-credential"))
-        .is_some_and(|credential| !credential.is_empty())
+    query.is_some_and(|query| {
+        form_urlencoded::parse(query.as_bytes())
+            .find(|(name, _)| name.eq_ignore_ascii_case("x-amz-credential"))
+            .is_some_and(|(_, credential)| !credential.is_empty())
+    })
 }

 /// Verify request has AWS PreSign Version '2'
@@ -1007,6 +1019,98 @@ pub fn get_query_param<'a>(query: &'a str, param_name: &str) -> Option<&'a str>
    None
 }

+/// Parse the RustFS presigned PutObject size capability after authentication.
+///
+/// The query value is covered by SigV4 when it is present before presigning, but
+/// the signature does not assign any semantics to the extension. Keep parsing
+/// strict and only enable the capability for a verified SigV4 presigned request.
+pub(crate) fn parse_presigned_put_max_content_length(
+    header: &HeaderMap,
+    query: Option<&str>,
+    verified_presigned: bool,
+) -> S3Result<Option<u64>> {
+    let Some(query) = query else {
+        return Ok(None);
+    };
+
+    let mut value = None;
+    let mut decoded_query = Vec::new();
+    for (name, candidate) in form_urlencoded::parse(query.as_bytes()) {
+        decoded_query.push((name.to_string(), candidate.to_string()));
+        if name == RUSTFS_MAX_CONTENT_LENGTH_QUERY {
+            if value.is_some() {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidRequest,
+                    format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} must appear exactly once"),
+                ));
+            }
+            value = Some(candidate.into_owned());
+        } else if name.eq_ignore_ascii_case(RUSTFS_MAX_CONTENT_LENGTH_QUERY) {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                format!("query parameter name must be exactly {RUSTFS_MAX_CONTENT_LENGTH_QUERY}"),
+            ));
+        }
+    }
+
+    let Some(value) = value else {
+        return Ok(None);
+    };
+
+    let query_value = |wanted: &str| {
+        decoded_query
+            .iter()
+            .find(|(name, _)| name.eq_ignore_ascii_case(wanted))
+            .map(|(_, value)| value.as_str())
+    };
+    let is_complete_sigv4_query = [
+        ("x-amz-algorithm", "AWS4-HMAC-SHA256"),
+        ("x-amz-date", ""),
+        ("x-amz-expires", ""),
+        ("x-amz-signedheaders", ""),
+        ("x-amz-credential", ""),
+        ("x-amz-signature", ""),
+    ]
+    .into_iter()
+    .all(|(name, expected)| {
+        query_value(name).is_some_and(|value| !value.is_empty() && (expected.is_empty() || value == expected))
+    });
+    if !verified_presigned
+        || !is_complete_sigv4_query
+        || !matches!(get_request_auth_type_with_query(header, Some(query)), AuthType::Presigned)
+    {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} requires a SigV4 presigned request"),
+        ));
+    }
+
+    let limit = value.parse::<u64>().map_err(|_| {
+        S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} must be a non-negative 64-bit integer"),
+        )
+    })?;
+
+    Ok(Some(limit))
+}
+
+/// Reject the PutObject-only size capability when it appears on another
+/// operation. Callers must invoke this after request authentication has run.
+pub(crate) fn reject_presigned_put_max_content_length_for_other_operation(
+    header: &HeaderMap,
+    query: Option<&str>,
+    verified_presigned: bool,
+) -> S3Result<()> {
+    if parse_presigned_put_max_content_length(header, query, verified_presigned)?.is_some() {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
+        ));
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -1651,6 +1755,67 @@ mod tests {
        assert_eq!(result, Some("value=with=equals"));
    }

+    #[test]
+    fn presigned_put_max_content_length_requires_exactly_one_signed_query_value() {
+        let headers = HeaderMap::new();
+        let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature";
+
+        let query = format!("{signed_prefix}&x-rustfs-max-content-length=104857600");
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(&query), true).unwrap(),
+            Some(104_857_600)
+        );
+
+        let encoded_credential = query.replacen("X-Amz-Credential", "X%2DAmz-Credential", 1);
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(&encoded_credential), true).unwrap(),
+            Some(104_857_600)
+        );
+
+        let duplicate = format!("{query}&x-rustfs-max-content-length=1");
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(&duplicate), true)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+
+        let wrong_case = format!("{signed_prefix}&X-RustFS-Max-Content-Length=1");
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(&wrong_case), true)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+
+        assert_eq!(
+            reject_presigned_put_max_content_length_for_other_operation(&headers, Some(&query), true)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+    }
+
+    #[test]
+    fn presigned_put_max_content_length_rejects_unsigned_or_invalid_values() {
+        let headers = HeaderMap::new();
+        let forged = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/credential&X-Amz-Signature=fake&x-rustfs-max-content-length=1";
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(forged), false)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+        for query in [
+            "x-rustfs-max-content-length=1",
+            "X-Amz-Credential=test/credential&x-rustfs-max-content-length=-1",
+            "X-Amz-Credential=test/credential&x-rustfs-max-content-length=18446744073709551616",
+        ] {
+            let error = parse_presigned_put_max_content_length(&headers, Some(query), true).unwrap_err();
+            assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
+        }
+    }
+
    #[test]
    fn test_credentials_is_expired() {
        let mut cred = create_test_credentials();
@@ -591,12 +591,19 @@ struct FeatureSpec {
    default_enabled: bool,
 }

-fn feature_specs() -> [FeatureSpec; 7] {
-    [
+fn feature_specs() -> &'static [FeatureSpec] {
+    &[
+        FeatureSpec {
+            name: "default",
+            enabled: cfg!(feature = "default"),
+            description: "Default feature set",
+            dependencies: "ftps + webdav",
+            default_enabled: true,
+        },
        FeatureSpec {
            name: "metrics-gpu",
            enabled: cfg!(feature = "metrics-gpu"),
-            description: "Metrics GPU support",
+            description: "GPU metrics support",
            dependencies: "rustfs-obs/gpu",
            default_enabled: false,
        },
@@ -610,7 +617,7 @@ fn feature_specs() -> [FeatureSpec; 7] {
        FeatureSpec {
            name: "swift",
            enabled: cfg!(feature = "swift"),
-            description: "Swift storage backend",
+            description: "OpenStack Swift protocol support",
            dependencies: "rustfs-protocols/swift",
            default_enabled: false,
        },
@@ -621,6 +628,13 @@ fn feature_specs() -> [FeatureSpec; 7] {
            dependencies: "rustfs-protocols/webdav",
            default_enabled: true,
        },
+        FeatureSpec {
+            name: "sftp",
+            enabled: cfg!(feature = "sftp"),
+            description: "SFTP protocol support",
+            dependencies: "rustfs-protocols/sftp",
+            default_enabled: false,
+        },
        FeatureSpec {
            name: "license",
            enabled: cfg!(feature = "license"),
@@ -635,11 +649,81 @@ fn feature_specs() -> [FeatureSpec; 7] {
            dependencies: "(none)",
            default_enabled: false,
        },
+        FeatureSpec {
+            name: "tracing-chunk-debug",
+            enabled: cfg!(feature = "tracing-chunk-debug"),
+            description: "Per-chunk data-plane tracing",
+            dependencies: "(none)",
+            default_enabled: false,
+        },
        FeatureSpec {
            name: "full",
            enabled: cfg!(feature = "full"),
-            description: "All features enabled",
-            dependencies: "metrics-gpu + ftps + swift + webdav",
+            description: "Full protocol and observability bundle",
+            dependencies: "metrics-gpu + ftps + swift + webdav + sftp + pyroscope",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "e2e-test-hooks",
+            enabled: cfg!(feature = "e2e-test-hooks"),
+            description: "End-to-end test hooks",
+            dependencies: "(none)",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "connect-e2e-short-credentials",
+            enabled: cfg!(feature = "connect-e2e-short-credentials"),
+            description: "Short-lived Connect credentials for debug E2E builds",
+            dependencies: "(none)",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "offline-enrollment-e2e-root",
+            enabled: cfg!(feature = "offline-enrollment-e2e-root"),
+            description: "Dedicated offline enrollment E2E root",
+            dependencies: "(none)",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "rio-v2",
+            enabled: cfg!(feature = "rio-v2"),
+            description: "RIO v2 storage path support",
+            dependencies: "rustfs-ecstore/rio-v2",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "pyroscope",
+            enabled: cfg!(feature = "pyroscope"),
+            description: "Pyroscope profiling support",
+            dependencies: "rustfs-obs/pyroscope",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "dial9",
+            enabled: cfg!(feature = "dial9"),
+            description: "Tokio runtime telemetry",
+            dependencies: "rustfs-obs/dial9",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "hotpath",
+            enabled: cfg!(feature = "hotpath"),
+            description: "Hotpath instrumentation",
+            dependencies: "hotpath + RustFS crate hotpath features",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "hotpath-alloc",
+            enabled: cfg!(feature = "hotpath-alloc"),
+            description: "Hotpath allocation diagnostics",
+            dependencies: "hotpath + hotpath/hotpath-alloc + RustFS crate hotpath-alloc features",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "hotpath-cpu",
+            enabled: cfg!(feature = "hotpath-cpu"),
+            description: "Hotpath CPU attribution",
+            dependencies: "hotpath + hotpath/hotpath-cpu + RustFS crate hotpath-cpu features",
            default_enabled: false,
        },
    ]
@@ -655,7 +739,7 @@ struct DepsInfoJson {

 fn collect_deps_info_json() -> DepsInfoJson {
    let features: Vec<FeatureInfoJson> = feature_specs()
-        .into_iter()
+        .iter()
        .map(|feature| FeatureInfoJson {
            name: feature.name,
            enabled: feature.enabled,
@@ -901,7 +985,7 @@ fn format_deps_info() -> String {
    output.push_str("### Feature Status\n\n");
    output.push_str("| Feature | Status | Description |\n");
    output.push_str("|---------|--------|-------------|\n");
-    for feature in &features {
+    for feature in features {
        let status = if feature.enabled { "✓" } else { "✗" };
        output.push_str(&format!("| {} | {} | {} |\n", feature.name, status, feature.description));
    }
@@ -916,7 +1000,7 @@ fn format_deps_info() -> String {
    output.push_str("\n### Feature Dependencies\n\n");
    output.push_str("| Feature | Dependencies |\n");
    output.push_str("|---------|-------------|\n");
-    for feature in &features {
+    for feature in features {
        output.push_str(&format!("| {} | {} |\n", feature.name, feature.dependencies));
    }

@@ -1001,10 +1085,22 @@ mod tests {
        let info = collect_deps_info_json();
        let feature_names: Vec<_> = info.features.iter().map(|feature| feature.name).collect();

-        assert_eq!(info.total_count, 7);
-        assert_eq!(info.features.len(), 7);
+        assert_eq!(info.total_count, 19);
+        assert_eq!(info.features.len(), 19);
+        assert!(feature_names.contains(&"default"));
        assert!(feature_names.contains(&"metrics-gpu"));
+        assert!(feature_names.contains(&"sftp"));
        assert!(feature_names.contains(&"io-scheduler-debug"));
+        assert!(feature_names.contains(&"tracing-chunk-debug"));
+        assert!(feature_names.contains(&"e2e-test-hooks"));
+        assert!(feature_names.contains(&"connect-e2e-short-credentials"));
+        assert!(feature_names.contains(&"offline-enrollment-e2e-root"));
+        assert!(feature_names.contains(&"rio-v2"));
+        assert!(feature_names.contains(&"pyroscope"));
+        assert!(feature_names.contains(&"dial9"));
+        assert!(feature_names.contains(&"hotpath"));
+        assert!(feature_names.contains(&"hotpath-alloc"));
+        assert!(feature_names.contains(&"hotpath-cpu"));
        assert!(!feature_names.contains(&"manual-test-runners"));
        assert!(!feature_names.contains(&"metrics"));
        assert!(!feature_names.contains(&"direct-io"));
@@ -1016,10 +1112,16 @@ mod tests {

        assert!(output.contains("| metrics-gpu |"));
        assert!(output.contains("| io-scheduler-debug |"));
+        assert!(output.contains("| tracing-chunk-debug |"));
+        assert!(output.contains("| sftp |"));
+        assert!(output.contains("| rio-v2 |"));
+        assert!(output.contains("| dial9 |"));
+        assert!(output.contains("| hotpath-cpu |"));
+        assert!(output.contains("| default | enabled by default |"));
        assert!(!output.contains("| manual-test-runners |"));
        assert!(output.contains("| ftps | enabled by default |"));
        assert!(output.contains("| webdav | enabled by default |"));
-        assert!(output.contains("| full | metrics-gpu + ftps + swift + webdav |"));
+        assert!(output.contains("| full | metrics-gpu + ftps + swift + webdav + sftp + pyroscope |"));
        assert!(!output.contains("| direct-io |"));
    }

@@ -17,6 +17,23 @@ use crate::storage_api::error::{QuotaError, StorageError};
 use rustfs_kms::KmsUnavailableError;
 use s3s::{S3Error, S3ErrorCode};

+/// Marks a request body that exceeded a presigned upload size capability.
+///
+/// This marker must survive the body-reader and storage layers so the client
+/// receives `EntityTooLarge` instead of a generic internal error.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct UploadLimitExceeded {
+    pub limit: u64,
+}
+
+impl std::fmt::Display for UploadLimitExceeded {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "upload exceeds the maximum content length of {} bytes", self.limit)
+    }
+}
+
+impl std::error::Error for UploadLimitExceeded {}
+
 #[derive(Debug)]
 pub struct ApiError {
    pub code: S3ErrorCode,
@@ -274,6 +291,17 @@ impl From<StorageError> for ApiError {
            };
        }

+        if let StorageError::Io(ref io_err) = err
+            && let Some(inner) = io_err.get_ref()
+            && error_chain_has_type::<UploadLimitExceeded>(inner)
+        {
+            return ApiError {
+                code: S3ErrorCode::EntityTooLarge,
+                message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
+                source: Some(Box::new(err)),
+            };
+        }
+
        if let StorageError::Io(ref io_err) = err
            && io_err
                .get_ref()
@@ -399,6 +427,13 @@ impl From<std::io::Error> for ApiError {
                    source: Some(Box::new(err)),
                };
            }
+            if error_chain_has_type::<UploadLimitExceeded>(inner) {
+                return ApiError {
+                    code: S3ErrorCode::EntityTooLarge,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
+                    source: Some(Box::new(err)),
+                };
+            }
            if error_chain_has_type::<rustfs_rio::IncompleteBody>(inner) {
                return ApiError {
                    code: S3ErrorCode::IncompleteBody,
@@ -815,6 +850,16 @@ mod tests {
        assert!(api_error.source.is_some());
    }

+    #[test]
+    fn upload_limit_marker_maps_to_entity_too_large_across_io_boundaries() {
+        let direct: ApiError = IoError::other(UploadLimitExceeded { limit: 5 }).into();
+        assert_eq!(direct.code, S3ErrorCode::EntityTooLarge);
+
+        let storage: ApiError = StorageError::Io(IoError::other(IoError::other(UploadLimitExceeded { limit: 5 }))).into();
+        assert_eq!(storage.code, S3ErrorCode::EntityTooLarge);
+        assert_eq!(storage.message, ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge));
+    }
+
    #[test]
    fn test_api_error_from_storage_io_copy_object_terminal_error_stays_internal() {
        let io_error = IoError::other(StorageError::FileCorrupt);
@@ -321,13 +321,15 @@ pub(crate) fn build_health_response_parts(
            ),
        };

-    let object_traffic_stalled = degraded_reasons.iter().any(|reason| {
+    let readiness_overlay_degraded = degraded_reasons.iter().any(|reason| {
        matches!(
            reason,
-            ReadinessDegradedReason::ObjectReadStalled | ReadinessDegradedReason::ObjectWriteStalled
+            ReadinessDegradedReason::ObjectReadStalled
+                | ReadinessDegradedReason::ObjectWriteStalled
+                | ReadinessDegradedReason::StartupFinalizationPending
        )
    });
-    if probe == HealthProbe::Readiness && (object_traffic_stalled || matches!(kms_ready, Some(false))) {
+    if probe == HealthProbe::Readiness && (readiness_overlay_degraded || matches!(kms_ready, Some(false))) {
        health = HealthCheckState {
            status_code: StatusCode::SERVICE_UNAVAILABLE,
            status: "degraded",
@@ -156,6 +156,7 @@ fn rustfs_s3_config() -> S3Config {
    let mut s3_config = S3Config::default();
    s3_config.normalize_forward_slash_path = true;
    s3_config.enable_sig_v2 = true;
+    s3_config.sig_v4_allowed_services.push("s3tables".to_string());
    s3_config
 }

@@ -1677,7 +1678,10 @@ fn process_connection(
                .option_layer(if is_console { Some(RedirectLayer) } else { None })
                .layer(BodylessStatusFixLayer)
                .layer(HeadRequestBodyFixLayer)
-                .layer(PublicHealthEndpointLayer::new(Arc::clone(&server_ctx)))
+                .layer(PublicHealthEndpointLayer::new(
+                    Arc::clone(&server_ctx),
+                    Arc::clone(&readiness),
+                ))
                .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer))
                .layer(DoubleSlashListBucketsCompatLayer)
                .service(service)
@@ -2270,6 +2274,9 @@ mod tests {

        assert!(s3_config.normalize_forward_slash_path);
        assert!(s3_config.enable_sig_v2);
+        assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "s3"));
+        assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "sts"));
+        assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "s3tables"));
    }

    #[test]
@@ -26,6 +26,7 @@ use crate::server::{
    build_health_response_parts, collect_probe_readiness, has_path_prefix, is_admin_path, is_table_catalog_path,
    kms_probe_staleness_limit, kms_ready_from_probe,
 };
+use crate::shared_types::ReadinessDegradedReason;
 use crate::storage_api::server::layer::apply_cors_headers;
 use crate::storage_api::server::layer::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced};
 use bytes::{Bytes, BytesMut};
@@ -36,6 +37,7 @@ use http_body_util::{BodyExt, Full};
 use hyper::body::Incoming;
 use pin_project_lite::pin_project;
 use quick_xml::events::Event;
+use rustfs_common::GlobalReadiness;
 use rustfs_obs::HTTP_SERVER_LOG_TARGET;
 #[cfg(feature = "swift")]
 use rustfs_protocols::swift::SwiftRouter;
@@ -1243,11 +1245,12 @@ where
 #[derive(Clone)]
 pub struct PublicHealthEndpointLayer {
    server_ctx: Arc<crate::runtime_sources::ServerContextSlot>,
+    readiness: Arc<GlobalReadiness>,
 }

 impl PublicHealthEndpointLayer {
-    pub fn new(server_ctx: Arc<crate::runtime_sources::ServerContextSlot>) -> Self {
-        Self { server_ctx }
+    pub fn new(server_ctx: Arc<crate::runtime_sources::ServerContextSlot>, readiness: Arc<GlobalReadiness>) -> Self {
+        Self { server_ctx, readiness }
    }
 }

@@ -1258,6 +1261,7 @@ impl<S> Layer<S> for PublicHealthEndpointLayer {
        PublicHealthEndpointService {
            inner,
            server_ctx: Arc::clone(&self.server_ctx),
+            readiness: Arc::clone(&self.readiness),
        }
    }
 }
@@ -1266,6 +1270,7 @@ impl<S> Layer<S> for PublicHealthEndpointLayer {
 pub struct PublicHealthEndpointService<S> {
    inner: S,
    server_ctx: Arc<crate::runtime_sources::ServerContextSlot>,
+    readiness: Arc<GlobalReadiness>,
 }

 fn health_endpoint_enabled() -> bool {
@@ -1334,6 +1339,7 @@ async fn build_public_health_http_response<RestBody, GrpcBody>(
    method: Method,
    path: String,
    object_traffic_health: Option<Arc<ObjectTrafficHealth>>,
+    readiness: &GlobalReadiness,
 ) -> Response<HybridBody<RestBody, GrpcBody>>
 where
    RestBody: From<Bytes>,
@@ -1358,7 +1364,15 @@ where
            .expect("failed to build health busy response");
    }

-    let readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await;
+    let mut readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await;
+    if probe == HealthProbe::Readiness
+        && !readiness.is_ready()
+        && let Some(report) = readiness_report.as_mut()
+    {
+        report
+            .degraded_reasons
+            .push(ReadinessDegradedReason::StartupFinalizationPending);
+    }
    let kms_ready = if probe == HealthProbe::Readiness && health_compat_kms_ready_check_enabled() {
        Some(health_kms_ready().await)
    } else {
@@ -1408,7 +1422,10 @@ where
                .server_ctx
                .installed_app_context()
                .map(|context| context.object_traffic_health());
-            return Box::pin(async move { Ok(build_public_health_http_response(method, path, object_traffic_health).await) });
+            let readiness = Arc::clone(&self.readiness);
+            return Box::pin(async move {
+                Ok(build_public_health_http_response(method, path, object_traffic_health, readiness.as_ref()).await)
+            });
        }

        let mut inner = self.inner.clone();
@@ -2210,14 +2227,25 @@ mod tests {
    use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};

    fn public_health_layer() -> PublicHealthEndpointLayer {
-        PublicHealthEndpointLayer::new(crate::runtime_sources::ServerContextSlot::new())
+        let readiness = Arc::new(GlobalReadiness::new());
+        readiness.mark_stage(rustfs_common::SystemStage::FullReady);
+        PublicHealthEndpointLayer::new(crate::runtime_sources::ServerContextSlot::new(), readiness)
    }

    async fn public_health_layer_with_tracker(object_traffic_health: Arc<ObjectTrafficHealth>) -> PublicHealthEndpointLayer {
+        let readiness = Arc::new(GlobalReadiness::new());
+        readiness.mark_stage(rustfs_common::SystemStage::FullReady);
+        public_health_layer_with_tracker_and_readiness(object_traffic_health, readiness).await
+    }
+
+    async fn public_health_layer_with_tracker_and_readiness(
+        object_traffic_health: Arc<ObjectTrafficHealth>,
+        readiness: Arc<GlobalReadiness>,
+    ) -> PublicHealthEndpointLayer {
        let app_context = crate::app::gating_test_env::app_context_with_object_traffic_health(object_traffic_health).await;
        let server_ctx = crate::runtime_sources::ServerContextSlot::new();
        assert!(server_ctx.install(app_context));
-        PublicHealthEndpointLayer::new(server_ctx)
+        PublicHealthEndpointLayer::new(server_ctx, readiness)
    }

    #[derive(Clone, Debug)]
@@ -2987,6 +3015,82 @@ mod tests {
        .await;
    }

+    #[tokio::test]
+    #[serial]
+    async fn public_readiness_waits_for_s3_admission_publication() {
+        async_with_vars(
+            [
+                (rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true")),
+                (rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false")),
+            ],
+            async {
+                let object_traffic_health = Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO));
+                let readiness = Arc::new(GlobalReadiness::new());
+                let inner = CountingHybridService::default();
+                let calls = inner.calls();
+                let mut service = public_health_layer_with_tracker_and_readiness(object_traffic_health, Arc::clone(&readiness))
+                    .await
+                    .layer(inner);
+
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::GET)
+                            .uri(HEALTH_READY_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("readiness request before admission publication"),
+                    )
+                    .await
+                    .expect("readiness response before admission publication");
+                assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+                let body = BodyExt::collect(response.into_body())
+                    .await
+                    .expect("readiness body before admission publication")
+                    .to_bytes();
+                let payload: serde_json::Value = serde_json::from_slice(&body).expect("readiness JSON");
+                assert_eq!(payload["ready"], false);
+                assert_eq!(payload["details"]["storage"]["ready"], true);
+                assert_eq!(payload["details"]["iam"]["ready"], true);
+                assert_eq!(payload["details"]["lock"]["ready"], true);
+                assert_eq!(payload["degradedReasons"], serde_json::json!(["startup_finalization_pending"]));
+
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::GET)
+                            .uri(HEALTH_COMPAT_LIVE_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("liveness request before admission publication"),
+                    )
+                    .await
+                    .expect("liveness response before admission publication");
+                assert_eq!(response.status(), StatusCode::OK);
+                let body = BodyExt::collect(response.into_body())
+                    .await
+                    .expect("liveness body before admission publication")
+                    .to_bytes();
+                let payload: serde_json::Value = serde_json::from_slice(&body).expect("liveness JSON");
+                assert_eq!(payload["status"], "ok");
+                assert!(payload.get("ready").is_none());
+
+                readiness.mark_stage(rustfs_common::SystemStage::FullReady);
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::HEAD)
+                            .uri(MINIO_HEALTH_READY_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("readiness request after admission publication"),
+                    )
+                    .await
+                    .expect("readiness response after admission publication");
+                assert_eq!(response.status(), StatusCode::OK);
+                assert_eq!(calls.load(Ordering::SeqCst), 0);
+            },
+        )
+        .await;
+    }
+
    #[tokio::test]
    #[serial]
    async fn public_readiness_aliases_use_the_installed_object_progress() {
@@ -45,6 +45,7 @@ pub enum ReadinessDegradedReason {
    ObjectWriteStalled,
    ClusterHealthTimeout,
    PeerHealthUnavailable,
+    StartupFinalizationPending,
    StorageAndIamUnavailable,
    StorageAndLockUnavailable,
    IamAndLockUnavailable,
@@ -62,6 +63,7 @@ impl ReadinessDegradedReason {
            ReadinessDegradedReason::ObjectWriteStalled => "object_write_stalled",
            ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout",
            ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable",
+            ReadinessDegradedReason::StartupFinalizationPending => "startup_finalization_pending",
            ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable",
            ReadinessDegradedReason::StorageAndLockUnavailable => "storage_and_lock_unavailable",
            ReadinessDegradedReason::IamAndLockUnavailable => "iam_and_lock_unavailable",
@@ -34,6 +34,9 @@ pub(crate) mod retry;
 pub(crate) mod state;
 pub(crate) mod transport;

+#[cfg(test)]
+mod tests;
+
 pub(crate) use self::hooks::*;
 pub(crate) use self::repair::*;
 pub(crate) use self::retry::*;
@@ -16,8 +16,9 @@ use super::ObjectOptions;
 use super::ecfs::FS;
 use super::{ECStore, PolicySys, ReplicationStatusType, StorageError, get_lock_acquire_timeout, is_err_bucket_not_found};
 use crate::auth::{
-    check_key_valid_with_context, get_condition_values_with_client_info, get_condition_values_with_query_and_client_info,
-    get_session_token,
+    AuthType, RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, check_key_valid_with_context,
+    get_condition_values_with_client_info, get_condition_values_with_query_and_client_info, get_request_auth_type_with_query,
+    get_session_token, parse_presigned_put_max_content_length,
 };
 use crate::error::ApiError;
 use crate::license::license_check;
@@ -1770,9 +1771,28 @@ impl S3Access for FS {

        // Publish this server's context slot so downstream data-plane handlers
        // resolve the same store (backlog#1052 S6).
-        let ext = cx.extensions_mut();
-        ext.insert(self.server_ctx().clone());
-        ext.insert(req_info);
+        let verified_presigned = matches!(get_request_auth_type_with_query(cx.headers(), cx.uri().query()), AuthType::Presigned);
+        {
+            let ext = cx.extensions_mut();
+            ext.insert(self.server_ctx().clone());
+            ext.insert(req_info);
+            if verified_presigned {
+                ext.insert(VerifiedPresignedRequest);
+            }
+        }
+
+        // The size capability is intentionally scoped to the single-object
+        // PutObject operation. Validate this at the operation-aware access
+        // boundary so unsupported GET/HEAD/DELETE/bucket routes cannot silently
+        // ignore a signed capability query.
+        if parse_presigned_put_max_content_length(cx.headers(), cx.uri().query(), verified_presigned)?.is_some()
+            && cx.s3_op().name() != "PutObject"
+        {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
+            ));
+        }
        license_check().map_err(|er| match er.kind() {
            std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"),
            _ => {
@@ -227,6 +227,8 @@ pub(crate) mod site_replication {
        BUCKET_REPLICATION_CONFIG, BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, BucketMetadata,
    };

+    #[cfg(test)]
+    pub(crate) use crate::storage::storage_api::ecstore_bucket::replication::merge_incoming_replication_config;
    pub(crate) use crate::storage::storage_api::ecstore_bucket::replication::{
        OperatorRuleContract, assign_site_replication_rule_priorities, is_site_replication_role,
        replication_target_arn_deployment_id, site_replication_rule_deployment_id,
@@ -238,6 +240,8 @@ pub(crate) mod site_replication {
    pub(crate) use crate::storage::storage_api::ecstore_bucket::versioning::VersioningApi;
    #[cfg(test)]
    pub(crate) use crate::storage::storage_api::ecstore_config::com::save_config;
+    #[cfg(test)]
+    pub(crate) use crate::storage::storage_api::{Endpoint, Endpoints, PoolEndpoints};

    pub(crate) use crate::storage::storage_api::{
        ECStore, EndpointServerPools, StorageError, delete_config_no_lock, lock_bucket_targets_metadata, read_config,
@@ -260,6 +264,8 @@ pub(crate) mod site_replication {
            LifecycleRule, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
            ReplicationRuleStatus, SourceSelectionCriteria, VersioningConfiguration,
        };
+        #[cfg(test)]
+        pub(crate) use s3s::dto::{ExpirationStatus, LifecycleExpiration, Timestamp, Transition, TransitionStorageClass};
        pub(crate) use s3s::{Body, S3Error, S3ErrorCode, S3Response, S3Result, s3_error};
    }
 }
@@ -88,7 +88,7 @@ PATTERNS=(
 # the guard fire again. Entries that stop matching anything are reported as
 # stale, so the list cannot decay into a blanket exclusion.
 #
-# 1-2: rustfs/src/admin/handlers/site_replication.rs negative fixtures for
+# 1-2: rustfs/src/site_replication/tests.rs negative fixtures for
 #      `validate_peer_connection_inner`, which must reject a private key
 #      submitted where a peer CA certificate is expected. Asserting on the
 #      rejection requires the header in the input; the key bodies are the
@@ -37,6 +37,7 @@ checked_files=(
  "rustfs/src/site_replication/retry.rs"
  "rustfs/src/site_replication/repair.rs"
  "rustfs/src/site_replication/hooks.rs"
+  "rustfs/src/site_replication/tests.rs"
  "rustfs/src/admin/handlers/group.rs"
  "rustfs/src/admin/handlers/quota.rs"
  "rustfs/src/admin/handlers/rebalance.rs"
Author	SHA1	Message	Date
houseme	a814bf3ae3	merge: resolve conflicts with main branch - Merge origin/main into perf/fileinfo-optimization - Resolve conflicts in crates/ecstore/src/set_disk/ops/object.rs - Keep AHashMap import and main branch's detailed imports Co-Authored-By: heihutu <heihutu@gmail.com>	2026-08-27 16:03:35 +08:00
GatewayJ	d902ac4f34	fix(s3): accept s3tables SigV4 service (#6719 ) Signed-off-by: houseme <housemecn@gmail.com> Co-authored-by: houseme <housemecn@gmail.com>	2026-08-27 15:46:08 +08:00
Henry Guo	80d0c51389	fix(server): align readiness with S3 admission (#6728 ) Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>	2026-08-27 15:45:46 +08:00
唐小鸭	daeaf40e2c	test: deflake config snapshot, presigned tamper, and pool resume tests (#6721 ) * test(ecstore): decouple server config snapshot test from global defaults The final assertion of server_config_snapshot_serializes_read_modify_write_transactions compared the second snapshot against a fresh Config::new(). Config::new() reads the process-global DEFAULT_KVS OnceLock, which a sibling test in the same process can register mid-run (crate::config::init()), so the in-process run 'cargo test -p rustfs-ecstore --lib config::' failed while nextest's process-per-test isolation hid the coupling. Assert on the snapshot's raw bytes against the baseline blob instead, which is deterministic and matches the invariant under test: the second transaction observes the store unchanged by the first. * test: deflake presigned tamper helper and relocated-pool resume staging tamper_signature only remapped '0' and 'a', so a signature containing neither (about 1 in 5000) left the URI unchanged and tripped the helper's own guard assert in CI. Complement every hex digit (15 - v) instead: the map has no fixed point, so the tamper always changes the value while keeping length and hex shape. execute_get_object_resumes_from_relocated_pool_without_splicing_body staged the relocation by reading xl.meta from every source-pool disk, but a write-quorum commit legitimately leaves a lagging minority disk without the object directory (#6701) — the test already tolerates that gap when normalizing the upload pool, and CI suite IO load hit the same gap in the staging loop. Skip sourceless disks, carry the staged metadata path explicitly, and assert a write-quorum majority was staged.	2026-08-27 15:19:22 +08:00
Zhengchao An	7b17d46ca9	refactor(site-replication): move business tests next to the service module (#6716 ) * refactor(site-replication): move business tests next to the service module backlog#1840 PR5: 79 business-logic tests (plus 12 helpers, 6 of them small fixtures kept on both sides) move from the admin handler file's test module into rustfs/src/site_replication/tests.rs, next to the code they exercise: peer connection/TLS/DNS/egress validation, the peer client cache and payload wire contract, retry-queue classification/settlement/escalation/backoff, the repair state machine, bootstrap-plan construction, lifecycle expiry subsetting, bucket-target reconciliation, endpoint/identity normalization, and state serialization. The 149 tests that exercise the admin handlers, apply/reconcile paths, status/resync builders, and the four include_str! tripwires stay in rustfs/src/admin/handlers/site_replication.rs with their subjects (229 total conserved: 149 + 79 + 1). The issue's PR5 also called for converting the source-order tripwire at the old file's line 11339 into a behavior test; both adversarial review passes re-derived all four tripwires against the shrunken file and found them non-vacuous and byte-identical in the regions they guard (the handler bodies, which did not move), so they stay as source-text assertions. Supporting changes: the root facade's site_replication consumer gains cfg(test) re-exports (endpoint types, merge_incoming_replication_config, five lifecycle DTO types) so the relocated tests stay off the direct s3s/admin surfaces — including rewriting the one inline crate::admin BucketMetadata path a moved test carried over (review finding); tests.rs joins the logging-guardrail checked list; the embedded-secrets guard comment follows the validate_peer_connection_inner fixtures to their new file. Verified: cargo check -p rustfs --all-targets clean; cargo nextest run -p rustfs --lib 3856/3856 passed; relocated tests run under site_replication::tests::; make pre-commit green including the s3s footprint ratchet; logging and embedded-secrets guards green. Refs rustfs/backlog#1840 * style(site-replication): apply rustfmt import ordering	2026-08-27 15:17:16 +08:00
houseme	c006f84461	feat(info): report all rustfs features (#6722 ) * feat(info): report all rustfs features Co-Authored-By: heihutu <heihutu@gmail.com> * chore(deps): update s3s revision Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs): adapt dial9 telemetry API Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>	2026-08-27 14:45:50 +08:00
cxymds	9a1a15ca58	feat(s3): limit presigned PutObject content length (#6724 ) feat(s3): limit presigned put content length	2026-08-27 13:44:04 +08:00
Zhengchao An	4bbc1d5640	test(ecstore): wait for multipart rename tail epochs (#6723 )	2026-08-27 05:28:55 +00:00
houseme	d628a2f48b	feat(filemeta): fix tests for AHashMap adaptation - Import AHashMap in metacache.rs - Fix metacache_entry_with_mod_time to use AHashMap - Fix metacache_entry_with_erasure_versions to use AHashMap - Fix metacache_entry_single_version to use AHashMap - Fix make_file_info_with_metadata to convert HashMap to AHashMap - Fix object_part_info_strategy to use AHashMap for checksums - Fix file_info_strategy to use AHashMap for metadata - Fix legacy_version_body_round_trips_through_encode to use AHashMap All 260 tests pass. AHashMap adaptation complete. Refs: https://github.com/rustfs/backlog/issues/2005 Co-Authored-By: heihutu <heihutu@gmail.com>	2026-08-26 23:34:33 +08:00
houseme	2bd1d70729	feat(ecstore): complete AHashMap adaptation - Add ahash dependency to ecstore crate - Import AHashMap in object.rs and bucket_lifecycle_ops.rs - Update StaleMultipartUploadCandidate.metadata to use AHashMap - Update stale_upload_lifecycle_due to use generics - Update stale_upload_current_size_with_opts to use generics - Fix all .into() calls to use iter().collect() for AHashMap conversion - Fix user_defined assignments to use iter().collect() - Fix replacement_metadata to use AHashMap All compilation errors resolved. rustfs-ecstore now compiles successfully. Refs: https://github.com/rustfs/backlog/issues/2005 Co-Authored-By: heihutu <heihutu@gmail.com>	2026-08-26 22:18:20 +08:00
houseme	0b96a992d6	feat(ecstore): continue AHashMap adaptation (partial) - Update merge_replication_metadata_lww to use generics - Update restore_metadata_update_preserves_protected_metadata to use generics - Update has_encrypted_part_layout_marker to use generics - Update clean_metadata, clean_metadata_keys, remove_standard_storage_class to use generics - Update update_hash_quorum_metadata_map, update_hash_target_delete_marker_versions to use generics - Fix fi.metadata assignment to use .into() - Fix lookup call to use get method - Fix replacement_metadata to use AHashMap Note: There are still 14 compilation errors remaining in ecstore. Refs: https://github.com/rustfs/backlog/issues/2005 Co-Authored-By: heihutu <heihutu@gmail.com>	2026-08-26 21:56:51 +08:00
houseme	bbc96c43a2	feat(ecstore): update functions to support AHashMap (partial) - Update restore_operation_id_from_metadata to use generics - Update require_restore_operation_id to use generics - Update restore_commit_operation_id_from_metadata to use generics - Update should_persist_encryption_original_size to use generics - Update strip_internal_multipart_metadata to use generics - Update multipart_bucket_incarnation_id to use generics - Update multipart_bucket_incarnation_matches to use generics - Update validate_multipart_bucket_incarnation to use generics - Update tier_destination_id_from_metadata to use generics - Update get_raw_etag to use generics Note: This is a partial implementation. There are still compilation errors in ecstore that need to be fixed. Refs: https://github.com/rustfs/backlog/issues/2005 Co-Authored-By: heihutu <heihutu@gmail.com>	2026-08-26 21:45:04 +08:00
houseme	ec9aabcf00	feat(fileinfo): use AHashMap for metadata fields - Add ahash dependency to workspace, filemeta, and utils crates - Change FileInfo.metadata to AHashMap<String, String> - Change ObjectPartInfo.checksums to Option<AHashMap<String, String>> - Change MetaObjectV1.meta to AHashMap<String, String> - Change MetaObjectV1Part.checksums to Option<AHashMap<String, String>> - Change UniquePartChecksums to use AHashMap - Make metadata_compat functions generic over BuildHasher - Make get_internal_replication_state generic over BuildHasher This optimization replaces the standard library's SipHash with ahash, which provides 2-3x faster hashing for typical key types. Refs: https://github.com/rustfs/backlog/issues/2005 Co-Authored-By: heihutu <heihutu@gmail.com>	2026-08-26 21:36:43 +08:00