test(utils): allow ephemeral port reuse

Cargo.lock

@@ -278,7 +278,6 @@ checksum = "312c1ea69e5fe9966e0029fb95aca8790100b85aff4f0d3b00a9337c74069a9c"
 dependencies = [
  "bigdecimal",
  "bon",
- "crc32fast",
  "digest 0.11.3",
  "log",
  "miniz_oxide 0.9.1",
@@ -290,11 +289,9 @@ dependencies = [
  "serde",
  "serde_bytes",
  "serde_json",
- "snap",
  "strum",
  "thiserror 2.0.20",
  "uuid",
- "zstd",
 ]
 
 [[package]]
@@ -3764,7 +3761,7 @@ checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
 
 [[package]]
 name = "e2e_test"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "anyhow",
  "astral-tokio-tar",
@@ -9093,7 +9090,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -9203,7 +9200,6 @@ dependencies = [
  "serial_test",
  "sha2 0.11.0",
  "shadow-rs",
- "snap",
  "socket2",
  "subtle",
  "sysinfo",
@@ -9231,7 +9227,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-audit"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "const-str",
@@ -9254,7 +9250,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-checksums"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "base64-simd",
  "bytes",
@@ -9270,7 +9266,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-common"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "chrono",
  "hotpath",
@@ -9288,7 +9284,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-concurrency"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
  "insta",
@@ -9301,7 +9297,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-config"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "const-str",
  "hotpath",
@@ -9311,7 +9307,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-credentials"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "base64-simd",
  "hmac 0.13.0",
@@ -9325,7 +9321,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-crypto"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "aes-gcm",
  "argon2",
@@ -9346,7 +9342,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-data-usage"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
  "rmp-serde",
@@ -9356,7 +9352,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-ecstore"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "arc-swap",
  "async-channel",
@@ -9495,7 +9491,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-extension-schema"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
  "serde",
@@ -9505,7 +9501,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-filemeta"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "arc-swap",
  "byteorder",
@@ -9532,7 +9528,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-heal"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "base64 0.23.1",
@@ -9563,7 +9559,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-iam"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "arc-swap",
  "async-trait",
@@ -9604,7 +9600,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-io-core"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "bytes",
  "hotpath",
@@ -9617,7 +9613,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-io-metrics"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "criterion",
  "hotpath",
@@ -9681,7 +9677,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-keystone"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "bytes",
  "futures",
@@ -9708,7 +9704,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-kms"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -9757,7 +9753,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-lifecycle"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "hotpath",
@@ -9780,7 +9776,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-lock"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "compact_str",
@@ -9803,7 +9799,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-log-analyzer"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "chrono",
  "flate2",
@@ -9822,7 +9818,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-madmin"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
  "humantime",
@@ -9837,7 +9833,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-notify"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "arc-swap",
  "async-trait",
@@ -9872,7 +9868,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-object-capacity"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "criterion",
  "futures",
@@ -9892,7 +9888,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-object-data-cache"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "bytes",
  "criterion",
@@ -9909,7 +9905,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-obs"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "chrono",
  "crossbeam-channel",
@@ -9964,7 +9960,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-policy"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "base64-simd",
@@ -9995,7 +9991,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-protocols"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "astral-tokio-tar",
  "async-compression",
@@ -10057,7 +10053,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-protos"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "flatbuffers",
  "hotpath",
@@ -10081,7 +10077,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-replication"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "byteorder",
  "bytes",
@@ -10099,7 +10095,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-rio"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "aes-gcm",
  "arc-swap",
@@ -10137,7 +10133,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-rio-v2"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "aes-gcm",
  "bytes",
@@ -10160,7 +10156,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-s3-ops"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
  "rustfs-s3-types",
@@ -10168,7 +10164,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-s3-types"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
  "serde",
@@ -10177,7 +10173,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-s3select-api"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "bytes",
@@ -10207,7 +10203,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-s3select-query"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-recursion",
  "async-trait",
@@ -10226,7 +10222,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-scanner"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "bytes",
@@ -10266,7 +10262,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-security-governance"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
  "thiserror 2.0.20",
@@ -10274,7 +10270,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-signer"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "base64-simd",
  "bytes",
@@ -10292,7 +10288,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-storage-api"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "hotpath",
@@ -10307,7 +10303,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-targets"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "arc-swap",
  "async-nats",
@@ -10361,7 +10357,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-test-utils"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "hotpath",
  "rustfs-data-usage",
@@ -10377,7 +10373,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-tls-runtime"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "arc-swap",
  "hotpath",
@@ -10398,7 +10394,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-trusted-proxies"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "async-trait",
  "axum",
@@ -10435,7 +10431,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-utils"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "base64-simd",
  "blake2",
@@ -10477,7 +10473,7 @@ dependencies = [
 
 [[package]]
 name = "rustfs-zip"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 dependencies = [
  "astral-tokio-tar",
  "async-compression",

Cargo.toml

@@ -69,7 +69,7 @@ edition = "2024"
 license = "Apache-2.0"
 repository = "https://github.com/rustfs/rustfs"
 rust-version = "1.97.1"
-version = "1.0.0-rc.3"
+version = "1.0.0-rc.2"
 homepage = "https://rustfs.com"
 description = "RustFS is a high-performance distributed object storage software built using Rust, one of the most popular languages worldwide. "
 keywords = ["RustFS", "Minio", "object-storage", "filesystem", "s3"]
@@ -86,52 +86,52 @@ redundant_clone = "warn"
 
 [workspace.dependencies]
 # RustFS Internal Crates
-rustfs = { path = "./rustfs", version = "1.0.0-rc.3" }
+rustfs = { path = "./rustfs", version = "1.0.0-rc.2" }
-rustfs-heal = { path = "crates/heal", version = "1.0.0-rc.3" }
+rustfs-heal = { path = "crates/heal", version = "1.0.0-rc.2" }
-rustfs-audit = { path = "crates/audit", version = "1.0.0-rc.3" }
+rustfs-audit = { path = "crates/audit", version = "1.0.0-rc.2" }
-rustfs-checksums = { path = "crates/checksums", version = "1.0.0-rc.3" }
+rustfs-checksums = { path = "crates/checksums", version = "1.0.0-rc.2" }
-rustfs-common = { path = "crates/common", version = "1.0.0-rc.3" }
+rustfs-common = { path = "crates/common", version = "1.0.0-rc.2" }
-rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-rc.3" }
+rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-rc.2" }
-rustfs-config = { path = "./crates/config", version = "1.0.0-rc.3" }
+rustfs-config = { path = "./crates/config", version = "1.0.0-rc.2" }
-rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-rc.3" }
+rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-rc.2" }
-rustfs-credentials = { path = "crates/credentials", version = "1.0.0-rc.3" }
+rustfs-credentials = { path = "crates/credentials", version = "1.0.0-rc.2" }
-rustfs-crypto = { path = "crates/crypto", version = "1.0.0-rc.3" }
+rustfs-crypto = { path = "crates/crypto", version = "1.0.0-rc.2" }
-rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-rc.3" }
+rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-rc.2" }
-rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-rc.3" }
+rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-rc.2" }
-rustfs-iam = { path = "crates/iam", version = "1.0.0-rc.3" }
+rustfs-iam = { path = "crates/iam", version = "1.0.0-rc.2" }
-rustfs-keystone = { path = "crates/keystone", version = "1.0.0-rc.3" }
+rustfs-keystone = { path = "crates/keystone", version = "1.0.0-rc.2" }
-rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-rc.3" }
+rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-rc.2" }
-rustfs-kms = { path = "crates/kms", version = "1.0.0-rc.3" }
+rustfs-kms = { path = "crates/kms", version = "1.0.0-rc.2" }
-rustfs-lock = { path = "crates/lock", version = "1.0.0-rc.3" }
+rustfs-lock = { path = "crates/lock", version = "1.0.0-rc.2" }
-rustfs-madmin = { path = "crates/madmin", version = "1.0.0-rc.3" }
+rustfs-madmin = { path = "crates/madmin", version = "1.0.0-rc.2" }
-rustfs-notify = { path = "crates/notify", version = "1.0.0-rc.3" }
+rustfs-notify = { path = "crates/notify", version = "1.0.0-rc.2" }
-rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-rc.3" }
+rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-rc.2" }
-rustfs-io-core = { path = "crates/io-core", version = "1.0.0-rc.3" }
+rustfs-io-core = { path = "crates/io-core", version = "1.0.0-rc.2" }
-rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-rc.3" }
+rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-rc.2" }
-rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-rc.3", default-features = false }
+rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-rc.2", default-features = false }
-rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-rc.3" }
+rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-rc.2" }
-rustfs-obs = { path = "crates/obs", version = "1.0.0-rc.3" }
+rustfs-obs = { path = "crates/obs", version = "1.0.0-rc.2" }
-rustfs-policy = { path = "crates/policy", version = "1.0.0-rc.3" }
+rustfs-policy = { path = "crates/policy", version = "1.0.0-rc.2" }
-rustfs-protos = { path = "crates/protos", version = "1.0.0-rc.3" }
+rustfs-protos = { path = "crates/protos", version = "1.0.0-rc.2" }
-rustfs-protocols = { path = "crates/protocols", version = "1.0.0-rc.3" }
+rustfs-protocols = { path = "crates/protocols", version = "1.0.0-rc.2" }
-rustfs-replication = { path = "crates/replication", version = "1.0.0-rc.3" }
+rustfs-replication = { path = "crates/replication", version = "1.0.0-rc.2" }
-rustfs-rio = { path = "crates/rio", version = "1.0.0-rc.3" }
+rustfs-rio = { path = "crates/rio", version = "1.0.0-rc.2" }
-rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-rc.3" }
+rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-rc.2" }
-rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-rc.3" }
+rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-rc.2" }
-rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-rc.3" }
+rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-rc.2" }
-rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-rc.3" }
+rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-rc.2" }
-rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-rc.3" }
+rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-rc.2" }
-rustfs-scanner = { path = "crates/scanner", version = "1.0.0-rc.3" }
+rustfs-scanner = { path = "crates/scanner", version = "1.0.0-rc.2" }
-rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-rc.3" }
+rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-rc.2" }
-rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-rc.3" }
+rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-rc.2" }
-rustfs-signer = { path = "crates/signer", version = "1.0.0-rc.3" }
+rustfs-signer = { path = "crates/signer", version = "1.0.0-rc.2" }
-rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-rc.3" }
+rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-rc.2" }
-rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-rc.3" }
+rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-rc.2" }
-rustfs-targets = { path = "crates/targets", version = "1.0.0-rc.3" }
+rustfs-targets = { path = "crates/targets", version = "1.0.0-rc.2" }
-rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-rc.3" }
+rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-rc.2" }
-rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-rc.3" }
+rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-rc.2" }
-rustfs-utils = { path = "crates/utils", version = "1.0.0-rc.3" }
+rustfs-utils = { path = "crates/utils", version = "1.0.0-rc.2" }
-rustfs-zip = { path = "./crates/zip", version = "1.0.0-rc.3" }
+rustfs-zip = { path = "./crates/zip", version = "1.0.0-rc.2" }
 
 # Async Runtime and Networking
 async-channel = "2.5.0"
@@ -171,7 +171,7 @@ tower = { version = "0.5.3" }
 tower-http = { version = "0.7.0" }
 
 # Serialization and Data Formats
-apache-avro = { version = "0.22.0", features = ["snappy", "zstandard"] }
+apache-avro = "0.22.0"
 bytes = { version = "1.12.1" }
 bytesize = "2.7.0"
 byteorder = "1.5.0"

README.md

@@ -116,7 +116,7 @@ chown -R 10001:10001 data logs
 docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
 
 # Using specific version
-docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.3
+docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.2
 ```
 
 If you use [podman](https://github.com/containers/podman) instead of docker, you can install the RustFS with the below command

README_ZH.md

@@ -113,7 +113,7 @@ chown -R 10001:10001 data logs
 docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
 
 # 使用指定版本运行
-docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.3
+docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.2
 ```
 
 如果您通过绑定挂载启用 TLS 证书目录，也请用同样方式准备该目录：

crates/checksums/src/base64.rs

@@ -11,6 +11,7 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
+#![allow(dead_code)]
 
 use base64_simd::STANDARD;
 

crates/checksums/src/http.rs

@@ -38,10 +38,7 @@ pub const XXHASH_3_HEADER_NAME: &str = "x-amz-checksum-xxhash3";
 pub const XXHASH_64_HEADER_NAME: &str = "x-amz-checksum-xxhash64";
 pub const XXHASH_128_HEADER_NAME: &str = "x-amz-checksum-xxhash128";
 
-#[allow(
+#[allow(dead_code)]
-    dead_code,
-    reason = "Content-MD5 wire name, resolved by header_name() below and asserted by this crate's tests (backlog#1823)"
-)]
 pub(crate) static MD5_HEADER_NAME: &str = "content-md5";
 
 pub const CHECKSUM_ALGORITHMS_IN_PRIORITY_ORDER: [&str; 5] =

crates/checksums/src/lib.rs

@@ -476,19 +476,13 @@ impl Checksum for Xxhash64 {
     }
 }
 
+#[allow(dead_code)]
 #[derive(Debug, Default)]
-#[allow(
-    dead_code,
-    reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
-)]
 struct Md5 {
     hasher: md5::Md5,
 }
 
-#[allow(
+#[allow(dead_code)]
-    dead_code,
-    reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
-)]
 impl Md5 {
     fn update(&mut self, bytes: &[u8]) {
         use md5::Digest;

crates/e2e_test/src/fake_s3_target/mod.rs

@@ -76,18 +76,6 @@ const SOURCE_MTIME_HEADERS: [&str; 2] = ["x-rustfs-source-mtime", "x-minio-sourc
 const SOURCE_REPLICATION_REQUEST_HEADERS: [&str; 2] =
     ["x-rustfs-source-replication-request", "x-minio-source-replication-request"];
 const SOURCE_ETAG_HEADERS: [&str; 2] = ["x-rustfs-source-etag", "x-minio-source-etag"];
-const SOURCE_TAGGING_TIMESTAMP_HEADERS: [&str; 2] = [
-    "x-rustfs-source-replication-tagging-timestamp",
-    "x-minio-source-replication-tagging-timestamp",
-];
-const SOURCE_RETENTION_TIMESTAMP_HEADERS: [&str; 2] = [
-    "x-rustfs-source-replication-retention-timestamp",
-    "x-minio-source-replication-retention-timestamp",
-];
-const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [
-    "x-rustfs-source-replication-legalhold-timestamp",
-    "x-minio-source-replication-legalhold-timestamp",
-];
 const RESERVED_BUCKET_PREFIXES: [&str; 3] = ["xn--", "sthree-", "amzn-s3-demo-"];
 const RESERVED_BUCKET_SUFFIXES: [&str; 6] = ["-s3alias", "--ol-s3", ".mrap", "--x-s3", "--table-s3", "-an"];
 
@@ -130,25 +118,6 @@ pub enum FaultAction {
     WrongEtag,
 }
 
-/// Replication LWW timestamp headers observed on a request, journaled so
-/// sender-side tests can assert what a real target would receive.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct ReplicationTimestampHeaders {
-    pub tagging: Option<String>,
-    pub retention: Option<String>,
-    pub legalhold: Option<String>,
-}
-
-impl ReplicationTimestampHeaders {
-    fn from_headers(headers: &HeaderMap) -> Self {
-        Self {
-            tagging: header_value(headers, &SOURCE_TAGGING_TIMESTAMP_HEADERS).map(bounded_journal_value),
-            retention: header_value(headers, &SOURCE_RETENTION_TIMESTAMP_HEADERS).map(bounded_journal_value),
-            legalhold: header_value(headers, &SOURCE_LEGALHOLD_TIMESTAMP_HEADERS).map(bounded_journal_value),
-        }
-    }
-}
-
 /// Credential-free request metadata retained for deterministic assertions.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RequestRecord {
@@ -162,7 +131,6 @@ pub struct RequestRecord {
     pub part_number: Option<i32>,
     pub content_length: Option<u64>,
     pub consumed_bytes: Option<usize>,
-    pub replication_timestamps: ReplicationTimestampHeaders,
     pub fault: Option<FaultAction>,
 }
 
@@ -568,15 +536,7 @@ impl S3Access for FaultAccess {
             .get(CONTENT_LENGTH)
             .and_then(|value| value.to_str().ok())
             .and_then(|value| value.parse().ok());
-        let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers());
+        let fault = record_request(&self.control, operation, context.method().clone(), parsed, content_length);
-        let fault = record_request(
-            &self.control,
-            operation,
-            context.method().clone(),
-            parsed,
-            content_length,
-            replication_timestamps,
-        );
         if let Some(RequestFault {
             action: FaultAction::Status(status),
             ..
@@ -629,7 +589,6 @@ fn record_request(
     method: Method,
     parsed: ParsedRequest,
     content_length: Option<u64>,
-    replication_timestamps: ReplicationTimestampHeaders,
 ) -> Option<RequestFault> {
     let mut state = lock(control);
     let action = parsed
@@ -654,7 +613,6 @@ fn record_request(
         part_number: parsed.part_number,
         content_length,
         consumed_bytes: None,
-        replication_timestamps,
         fault: action.clone(),
     });
     action.map(|action| RequestFault { sequence, action })
@@ -1741,52 +1699,6 @@ mod tests {
             .await?)
     }
 
-    #[tokio::test]
-    async fn journals_replication_timestamp_headers() -> Result<(), BoxError> {
-        let target = FakeS3Target::start().await?;
-        target.create_bucket("target-bucket");
-        let client = client(&target);
-
-        client
-            .put_object()
-            .bucket("target-bucket")
-            .key("plain")
-            .body(ByteStream::from_static(b"plain"))
-            .send()
-            .await?;
-        client
-            .put_object()
-            .bucket("target-bucket")
-            .key("stamped")
-            .body(ByteStream::from_static(b"stamped"))
-            .customize()
-            .map_request(move |mut request| {
-                let headers = request.headers_mut();
-                headers.insert("x-rustfs-source-replication-tagging-timestamp", "2026-01-02T03:04:05Z");
-                headers.insert("x-minio-source-replication-retention-timestamp", "2026-01-02T03:04:06Z");
-                headers.insert("x-rustfs-source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z");
-                Ok::<_, std::convert::Infallible>(request)
-            })
-            .send()
-            .await?;
-
-        let requests = target.requests();
-        let plain = requests
-            .iter()
-            .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("plain"))
-            .expect("plain PUT must be journaled");
-        assert_eq!(plain.replication_timestamps, ReplicationTimestampHeaders::default());
-
-        let stamped = requests
-            .iter()
-            .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("stamped"))
-            .expect("stamped PUT must be journaled");
-        assert_eq!(stamped.replication_timestamps.tagging.as_deref(), Some("2026-01-02T03:04:05Z"));
-        assert_eq!(stamped.replication_timestamps.retention.as_deref(), Some("2026-01-02T03:04:06Z"));
-        assert_eq!(stamped.replication_timestamps.legalhold.as_deref(), Some("2026-01-02T03:04:07Z"));
-        Ok(())
-    }
-
     macro_rules! assert_sdk_error {
         ($error:expr, $status:expr, $code:expr) => {{
             let error = &$error;
@@ -3073,7 +2985,6 @@ mod tests {
                         part_number: None,
                     },
                     Some(0),
-                    ReplicationTimestampHeaders::default(),
                 );
             }
             let records = lock(&control).requests.clone();
@@ -3095,7 +3006,6 @@ mod tests {
                     part_number: None,
                 },
                 None,
-                ReplicationTimestampHeaders::default(),
             );
             {
                 let bounded_records = lock(&bounded_control);

crates/ecstore/src/api/mod.rs

@@ -135,8 +135,7 @@ pub mod bucket {
         pub use crate::bucket::metadata_sys::ConfigWriteLockProbe;
         pub use crate::bucket::metadata_sys::{
             BucketMetadataMutationGuard, BucketMetadataSys, ObjectLockConfigState, acquire_bucket_metadata_transaction_lock,
-            acquire_bucket_metadata_transaction_lock_for_incarnation, capture_bucket_metadata_incarnation, delete,
+            capture_bucket_metadata_incarnation, delete, delete_if_incarnation, get, get_accelerate_config, get_bucket_policy,
-            delete_if_incarnation, delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy,
             get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config,
             get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
             get_object_lock_config, get_object_lock_config_state, get_public_access_block_config, get_quota_config,
@@ -185,18 +184,17 @@ pub mod bucket {
             mrf_backlog_observability_snapshot,
         };
         pub use crate::bucket::replication::{
-            BucketReplicationResyncStatus, BucketReplicationStat, BucketReplicationStats, BucketStats,
+            BucketReplicationResyncStatus, BucketReplicationStats, BucketStats, DeleteReplicationConfigSnapshot,
-            DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, InQueueMetric,
+            DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, MrfOpKind, MrfReplicateEntry,
-            MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION,
+            MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS,
-            REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE,
+            REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION,
-            REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
+            REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo,
-            ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
+            ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigStructureError, ReplicationConfigurationExt,
-            ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationDeleteScheduleInput,
+            ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge,
-            ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
+            ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission,
-            ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
+            ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage,
-            ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
+            ReplicationTargetValidationError, ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog,
-            ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
+            TargetReplicationResyncStatus, VersionPurgeStatusType, commit_force_delete_intent, complete_force_delete_intent,
-            VersionPurgeStatusType, XferStats, commit_force_delete_intent, complete_force_delete_intent,
             delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
             get_global_replication_stats, init_background_replication, invalid_replication_config_status_field,
             persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta,

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -58,9 +58,7 @@ use rustfs_utils::http::{
 };
 use rustfs_utils::http::{
     SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
-    SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
+    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header,
-    SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
-    insert_header,
 };
 use rustls_pki_types::pem::PemObject;
 use serde::{Deserialize, Serialize};
@@ -82,6 +80,7 @@ use tracing::warn;
 use url::Url;
 use uuid::Uuid;
 
+const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60);
 const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
 const REDACTED_CREDENTIAL: &str = "<redacted>";
 
@@ -1477,12 +1476,9 @@ impl Default for AdvancedPutOptions {
             replication_status: ReplicationStatusType::Pending,
             source_mtime: OffsetDateTime::now_utc(),
             replication_request: false,
-            // UNIX_EPOCH means "never modified": header() must not emit a
+            retention_timestamp: OffsetDateTime::now_utc(),
-            // timestamp header for it, otherwise a receiver would treat an
+            tagging_timestamp: OffsetDateTime::now_utc(),
-            // unset category as a modification made right now.
+            legalhold_timestamp: OffsetDateTime::now_utc(),
-            retention_timestamp: OffsetDateTime::UNIX_EPOCH,
-            tagging_timestamp: OffsetDateTime::UNIX_EPOCH,
-            legalhold_timestamp: OffsetDateTime::UNIX_EPOCH,
             replication_validity_check: false,
         }
     }
@@ -1679,16 +1675,6 @@ impl PutObjectOptions {
             );
         }
 
-        for (suffix, timestamp) in [
-            (SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, self.internal.tagging_timestamp),
-            (SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, self.internal.retention_timestamp),
-            (SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, self.internal.legalhold_timestamp),
-        ] {
-            if timestamp.unix_timestamp() != 0 {
-                insert_header(&mut header, suffix, timestamp.format(&Rfc3339).unwrap_or_default());
-            }
-        }
-
         if self.internal.replication_request {
             insert_header(&mut header, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
         }
@@ -2856,57 +2842,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn put_object_headers_carry_replication_timestamp_headers() {
-        // MinIO receivers resolve concurrent tag/retention/legal-hold edits by
-        // last-writer-wins on these headers (object-api-options.go parses them
-        // as RFC3339); a replica without them loses every conflict resolution.
-        let mut opts = PutObjectOptions::default();
-        opts.internal.replication_request = true;
-        let tagging = OffsetDateTime::from_unix_timestamp(1_700_000_001).expect("valid timestamp");
-        let retention = OffsetDateTime::from_unix_timestamp(1_700_000_002).expect("valid timestamp");
-        let legalhold = OffsetDateTime::from_unix_timestamp(1_700_000_003).expect("valid timestamp");
-        opts.internal.tagging_timestamp = tagging;
-        opts.internal.retention_timestamp = retention;
-        opts.internal.legalhold_timestamp = legalhold;
-
-        let header = opts.header();
-        for (suffix, expected) in [
-            ("source-replication-tagging-timestamp", tagging),
-            ("source-replication-retention-timestamp", retention),
-            ("source-replication-legalhold-timestamp", legalhold),
-        ] {
-            assert_eq!(
-                rustfs_utils::http::get_header(&header, suffix).as_deref(),
-                Some(expected.format(&Rfc3339).expect("RFC3339 timestamp").as_str()),
-                "replication put requests must carry the {suffix} header"
-            );
-        }
-    }
-
-    #[test]
-    fn put_object_headers_omit_unset_replication_timestamps() {
-        // UNIX_EPOCH means "never modified on the source"; sending it would
-        // make the receiver treat an unset category as a fresh modification.
-        let mut opts = PutObjectOptions::default();
-        opts.internal.replication_request = true;
-        opts.internal.tagging_timestamp = OffsetDateTime::UNIX_EPOCH;
-        opts.internal.retention_timestamp = OffsetDateTime::UNIX_EPOCH;
-        opts.internal.legalhold_timestamp = OffsetDateTime::UNIX_EPOCH;
-
-        let header = opts.header();
-        for suffix in [
-            "source-replication-tagging-timestamp",
-            "source-replication-retention-timestamp",
-            "source-replication-legalhold-timestamp",
-        ] {
-            assert!(
-                rustfs_utils::http::get_header(&header, suffix).is_none(),
-                "unset {suffix} must not be sent to replication targets"
-            );
-        }
-    }
-
     #[tokio::test]
     async fn get_remote_target_client_internal_rejects_loopback_endpoint() {
         let sys = BucketTargetSys::default();

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -126,23 +126,11 @@ const EVENT_LIFECYCLE_EXPIRED_DETECTED: &str = "lifecycle_expired_detected";
 const EVENT_LIFECYCLE_NOT_ENQUEUED: &str = "lifecycle_not_enqueued";
 const EVENT_LIFECYCLE_DELETE_DISPATCHED: &str = "lifecycle_delete_dispatched";
 const EVENT_LIFECYCLE_DELETE_COMPLETED: &str = "lifecycle_delete_completed";
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 const EVENT_LIFECYCLE_TIER_AUDIT: &str = "lifecycle_tier_audit";
 const EVENT_LIFECYCLE_TIER_OPERATION_FAILED: &str = "lifecycle_tier_operation_failed";
 const EVENT_LIFECYCLE_DELETE_FAILED: &str = "lifecycle_delete_failed";
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub type TimeFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub type TraceFn =
     Arc<dyn Fn(String, HashMap<String, String>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
 pub type ExpiryOpType = Box<dyn ExpiryOp + Send + Sync + 'static>;
@@ -152,21 +140,9 @@ static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 
 pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count";
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub const AMZ_TAG_DIRECTIVE: &str = "X-Amz-Tagging-Directive";
 pub const AMZ_ENCRYPTION_AES: &str = "AES256";
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms";
 
 pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier.";
@@ -304,10 +280,6 @@ impl LifecycleSys {
         }
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-    )]
     pub fn trace(oi: &ObjectInfo) -> TraceFn {
         let bucket = oi.bucket.clone();
         let name = oi.name.clone();
@@ -598,10 +570,6 @@ async fn delete_free_version_remote_object(
     Ok(())
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 async fn delete_free_version_remote_object_then<T, F, Fut>(
     oi: &ObjectInfo,
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
@@ -2900,10 +2868,6 @@ fn stale_upload_default_due(initiated: OffsetDateTime, default_expiry: StdDurati
     initiated + time::Duration::seconds(default_expiry.as_secs() as i64)
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 async fn stale_upload_current_size(set: &Arc<SetDisks>, metadata: &HashMap<String, String>, upload_dir: &str) -> Option<usize> {
     stale_upload_current_size_with_opts(set, metadata, upload_dir, false).await
 }
@@ -3388,10 +3352,6 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu
     Ok(())
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 fn mark_delete_opts_skip_decommissioned_on_remote_success(opts: &mut ObjectOptions, remote_delete_succeeded: bool) {
     if remote_delete_succeeded {
         opts.skip_decommissioned = true;
@@ -4379,10 +4339,6 @@ pub async fn expire_transitioned_object(
     Ok(dobj)
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub fn gen_transition_objname(bucket: &str) -> Result<String, Error> {
     let us = Uuid::new_v4().to_string();
     let mut hasher = Sha256::new();
@@ -4417,10 +4373,6 @@ pub async fn transition_object(api: Arc<ECStore>, oi: &ObjectInfo, lae: LcAuditE
     result
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
     let tier = _tier.to_string();
     Arc::new(move || {
@@ -4439,10 +4391,6 @@ pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
     })
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn get_transitioned_object_reader(
     bucket: &str,
     object: &str,
@@ -5197,10 +5145,6 @@ async fn lifecycle_delete_config_snapshot(api: &ECStore, oi: &ObjectInfo) -> Res
     ReplicationObjectBridge::delete_request_config(api, &oi.bucket).await
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
     let mut success = false;
     match event.action {
@@ -7478,10 +7422,6 @@ mod tests {
     // process environment while `env::set_var`/`env::remove_var` is active.
     // SAFETY: keep this note adjacent to the allowance for the repository guard.
     #[allow(unsafe_code)]
-    #[allow(
-        dead_code,
-        reason = "transition-queue env fixture kept for tests that scope those vars; no test uses it today (backlog#1823)"
-    )]
     async fn with_transition_queue_env_async<F, Fut>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
     where
         F: FnOnce() -> Fut,

crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs

@@ -759,10 +759,6 @@ pub struct ManualTransitionWorkerResultRecord {
 }
 
 impl ManualTransitionWorkerResultRecord {
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-    )]
     pub fn new(job_id: Uuid, task_key: impl Into<String>, result: ManualTransitionWorkerResult) -> Self {
         Self::new_with_reason(job_id, task_key, result, None)
     }
@@ -1261,10 +1257,6 @@ pub(crate) async fn save_manual_transition_task_if_absent(
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn load_manual_transition_task_record(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -1328,10 +1320,6 @@ async fn scan_manual_transition_task_journal(api: Arc<ECStore>, job_id: Uuid) ->
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn load_manual_transition_worker_result_stats(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -1467,10 +1455,6 @@ async fn scan_manual_transition_worker_result_journal(
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn reconcile_manual_transition_worker_results(
     api: Arc<ECStore>,
     job_id: Uuid,

crates/ecstore/src/bucket/lifecycle/replication_sink.rs

@@ -15,35 +15,25 @@
 use rustfs_common::metrics::IlmAction;
 
 use crate::bucket::lifecycle::lifecycle::ObjectOpts;
-use crate::bucket::replication::ReplicationLifecycleBridge;
 pub(crate) use crate::bucket::replication::ReplicationStatusType;
 #[cfg(test)]
 pub(crate) use crate::bucket::replication::VersionPurgeStatusType;
 pub(crate) use crate::bucket::replication::{
     DeleteReplicationConfigSnapshot, ReplicationObjectBridge, replication_state_to_filemeta,
 };
+use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
 use crate::storage_api_contracts::object::DeletedObject;
 
-#[allow(
+pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig;
-    dead_code,
+
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) fn has_pending_version_purge(obj: &ObjectOpts) -> bool {
     obj.version_purge_status.is_pending()
 }
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) fn has_pending_object_replication(obj: &ObjectOpts) -> bool {
     replication_status_blocks_lifecycle(&obj.replication_status)
 }
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) fn has_pending_lifecycle_replication(obj: &ObjectOpts) -> bool {
     has_pending_object_replication(obj) || has_pending_version_purge(obj)
 }

crates/ecstore/src/bucket/lifecycle/tagging_boundary.rs

@@ -14,10 +14,6 @@
 
 use std::collections::HashMap;
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) fn decode_tags_to_map(tags: &str) -> HashMap<String, String> {
     crate::bucket::tagging::decode_tags_to_map(tags)
 }

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -331,10 +331,6 @@ where
     persist_tier_delete_journal_entry(api, &committed).await
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn abort_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
 where
     S: ObjectOperations<

crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs

@@ -148,10 +148,6 @@ struct RecoveryCursor {
     object: String,
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn recover_tier_free_versions(
     api: Arc<ECStore>,
     limit: usize,

crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs

@@ -385,10 +385,6 @@ impl ExpiryOp for Jentry {
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
     let result = delete_object_from_remote_tier_raw(obj_name, rv_id, tier_name).await;
     if let Err(err) = &result
@@ -399,10 +395,6 @@ pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_na
     result
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
     #[cfg(test)]
     if let Some(result) = run_remote_tier_delete_test_hook(obj_name, rv_id, tier_name) {
@@ -413,10 +405,6 @@ async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_na
     delete_object_from_remote_tier_raw_with_manager(obj_name, rv_id, tier_name, &tier_config_mgr).await
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 async fn delete_object_from_remote_tier_raw_with_manager(
     obj_name: &str,
     rv_id: &str,
@@ -497,10 +485,6 @@ pub enum RemoteTierDeleteOutcome {
     AlreadyRemoved,
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn delete_object_from_remote_tier_idempotent(
     obj_name: &str,
     rv_id: &str,

crates/ecstore/src/bucket/lifecycle/transition_transaction.rs

@@ -50,16 +50,8 @@ pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
 #[derive(Debug, thiserror::Error)]
 pub enum TransitionTransactionError {
     #[error("transition transaction already exists")]
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-    )]
     AlreadyExists,
     #[error("transition transaction is not found")]
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-    )]
     NotFound,
     #[error("transition transaction is corrupt: {0}")]
     Corrupt(&'static str),

crates/ecstore/src/bucket/metadata_sys.rs

@@ -60,14 +60,12 @@ struct ConfigWriteLockProbeState {
 static CONFIG_WRITE_LOCK_PROBES: std::sync::OnceLock<StdMutex<Vec<Arc<ConfigWriteLockProbeState>>>> = std::sync::OnceLock::new();
 
 #[cfg(any(test, feature = "test-util"))]
-#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
 pub struct ConfigWriteLockProbe {
     state: Arc<ConfigWriteLockProbeState>,
 }
 
 #[cfg(any(test, feature = "test-util"))]
 impl ConfigWriteLockProbe {
-    #[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
     pub fn install(bucket: &str) -> Self {
         let state = Arc::new(ConfigWriteLockProbeState {
             bucket: bucket.to_string(),
@@ -86,7 +84,6 @@ impl ConfigWriteLockProbe {
         Self { state }
     }
 
-    #[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
     pub async fn wait_until_attempted(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
@@ -659,16 +656,6 @@ pub async fn update_under_transaction_lock(
     update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await
 }
 
-/// Clear one config file while the caller holds this bucket's transaction lock.
-pub async fn delete_under_transaction_lock(
-    guard: &BucketMetadataMutationGuard,
-    bucket: &str,
-    config_file: &str,
-) -> Result<OffsetDateTime> {
-    guard.ensure_valid(bucket)?;
-    delete_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file).await
-}
-
 pub async fn update_quota_if_incarnation(
     bucket: &str,
     data: Vec<u8>,
@@ -808,14 +795,6 @@ pub async fn acquire_bucket_metadata_transaction_lock(bucket: &str) -> Result<Bu
     acquire_config_write_guard(get_bucket_metadata_sys()?, bucket).await
 }
 
-/// Acquire the bucket transaction lock only if its incarnation still matches.
-pub async fn acquire_bucket_metadata_transaction_lock_for_incarnation(
-    bucket: &str,
-    expected_incarnation_id: Uuid,
-) -> Result<BucketMetadataMutationGuard> {
-    acquire_config_write_guard_for_incarnation(get_bucket_metadata_sys()?, bucket, Some(expected_incarnation_id)).await
-}
-
 pub(crate) async fn acquire_bucket_metadata_transaction_lock_in(
     ctx: &crate::runtime::instance::InstanceContext,
     bucket: &str,
@@ -893,10 +872,6 @@ pub async fn get_bucket_policy_raw(bucket: &str) -> Result<(String, OffsetDateTi
     bucket_meta_sys.get_bucket_policy_raw(bucket).await
 }
 
-#[allow(
-    dead_code,
-    reason = "free-function facade over the live BucketMetadataSys::get_bucket_acl_config; no caller in this port (backlog#1823)"
-)]
 pub async fn get_bucket_acl_config(bucket: &str) -> Result<(String, OffsetDateTime)> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1111,10 +1086,6 @@ pub async fn get_config_from_disk(bucket: &str) -> Result<BucketMetadata> {
     bucket_meta_sys.get_config_from_disk(bucket).await
 }
 
-#[allow(
-    dead_code,
-    reason = "ambient-facade variant of the live created_at_in; no caller in this port (backlog#1823)"
-)]
 pub async fn created_at(bucket: &str) -> Result<OffsetDateTime> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1628,7 +1599,6 @@ impl BucketMetadataSys {
     /// [`Self::update`], with the payload computed from the loaded metadata
     /// instead of supplied up front. Loads through this system's own store so
     /// the read and the persisted write target the same instance.
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn update_config_with<F>(&self, bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
     where
         F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
@@ -1733,7 +1703,6 @@ impl BucketMetadataSys {
     /// A miss is never published as an authoritative default, and a snapshot
     /// read before delete plus same-name recreation cannot replace the new
     /// generation.
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
         if is_meta_bucketname(bucket) {
             return Err(Error::other("errInvalidArgument"));

crates/ecstore/src/bucket/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 // #730: bucket subsystems still contain staged ECStore migration code.
+#![allow(dead_code)]
 
 pub mod bandwidth;
 pub mod bucket_target_sys;

crates/ecstore/src/bucket/object_lock/objectlock_sys.rs

@@ -136,7 +136,6 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
 
 /// Check if an object has legal hold enabled.
 /// Returns true if legal hold is ON.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
     let lhold = objectlock::get_object_legalhold_meta(user_defined);
     matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
@@ -152,7 +151,6 @@ fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> b
 /// # Returns
 /// * `true` if the object is locked (cannot be deleted/modified)
 /// * `false` if the object is not locked
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_object_locked_by_metadata(user_defined: &std::collections::HashMap<String, String>, is_delete_marker: bool) -> bool {
     // Delete markers are never locked
     if is_delete_marker {

crates/ecstore/src/bucket/quota/mod.rs

@@ -193,7 +193,6 @@ pub enum QuotaError {
 }
 
 #[derive(Debug, Serialize)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub struct QuotaErrorResponse {
     #[serde(rename = "Code")]
     pub code: String,
@@ -209,7 +208,6 @@ pub struct QuotaErrorResponse {
 }
 
 impl QuotaErrorResponse {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn new(quota_error: &QuotaError, request_id: &str, host_id: &str) -> Self {
         match quota_error {
             QuotaError::QuotaExceeded { .. } => Self {

crates/ecstore/src/bucket/quota/reservation.rs

@@ -899,7 +899,6 @@ async fn save_ledger_locked(
 }
 
 #[cfg(any(test, feature = "test-util"))]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn fail_next_quota_ledger_save_for_test() {
     FAIL_NEXT_LEDGER_SAVE.store(true, std::sync::atomic::Ordering::SeqCst);
 }

crates/ecstore/src/bucket/replication/mod.rs

@@ -60,7 +60,7 @@ pub use replication_filemeta_boundary::{
 pub(crate) use replication_filemeta_boundary::{
     replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta,
 };
-pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge;
+pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
 pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
 pub use replication_object_bridge::ReplicationObjectBridge;
 pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};
@@ -81,6 +81,6 @@ pub use replication_queue_boundary::{
 pub use replication_resync_boundary::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus};
 pub use replication_scanner_bridge::ReplicationScannerBridge;
 pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog};
-pub use replication_stats_boundary::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};
+pub use replication_stats_boundary::{BucketReplicationStats, BucketStats};
 pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage};
 pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge;

crates/ecstore/src/bucket/replication/replication_config_store.rs

@@ -37,10 +37,6 @@ impl ReplicationConfigStore {
         com::read_config_limited(api, file, max_bytes).await
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     pub(crate) async fn read_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
     where
         S: ReplicationObjectIO,

crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs

@@ -24,27 +24,15 @@ use super::replication_storage_boundary::{
     DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, deleted_object_for_replication,
 };
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) type ReplicationLifecycleConfig = ReplicationConfig;
 
 pub(crate) struct ReplicationLifecycleBridge;
 
 impl ReplicationLifecycleBridge {
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn new_config(config: ReplicationConfiguration) -> ReplicationLifecycleConfig {
         ReplicationConfig::new(Some(config), None)
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn has_pending_version_purge(
         config: &ReplicationLifecycleConfig,
         object_name: &str,
@@ -57,10 +45,6 @@ impl ReplicationLifecycleBridge {
                 .is_some_and(|config| config.has_active_rules(object_name, true))
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) async fn check_delete_replication(
         bucket: &str,
         object: &ObjectToDelete,
@@ -70,10 +54,6 @@ impl ReplicationLifecycleBridge {
         check_replicate_delete(bucket, object, source, opts, None).await
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn version_delete_replication_state(decision: &ReplicateDecision) -> ReplicationState {
         let pending_status = decision.pending_status();
         ReplicationState {

crates/ecstore/src/bucket/replication/replication_msgp_boundary.rs

@@ -19,33 +19,17 @@ use time::OffsetDateTime;
 use super::replication_error_boundary::Result;
 use crate::bucket::msgp_decode;
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) struct ReplicationMsgpCodec;
 
 impl ReplicationMsgpCodec {
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn read_ext8_time<R: Read>(rd: &mut R) -> Result<OffsetDateTime> {
         msgp_decode::read_msgp_ext8_time(rd)
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn skip_value<R: Read>(rd: &mut R) -> Result<()> {
         msgp_decode::skip_msgp_value(rd)
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn write_time<W: Write>(wr: &mut W, time: OffsetDateTime) -> Result<()> {
         msgp_decode::write_msgp_time(wr, time)
     }

crates/ecstore/src/bucket/replication/replication_object_bridge.rs

@@ -77,10 +77,6 @@ impl ReplicationObjectBridge {
         load_delete_request_config_in(ctx, bucket).await
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) async fn delete_config_snapshot_in(
         ctx: &ReplicationInstanceContext,
         bucket: &str,

crates/ecstore/src/bucket/replication/replication_object_config.rs

@@ -231,10 +231,6 @@ pub(crate) async fn load_delete_replication_config(
     delete_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata(bucket).await?)
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-)]
 pub(crate) async fn load_delete_replication_config_in(
     ctx: &ReplicationInstanceContext,
     bucket: &str,

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -217,10 +217,6 @@ impl DurableMrfBacklogTracker {
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-)]
 fn durable_mrf_backlog_tracker_from_entries(entries: &[MrfReplicateEntry]) -> DurableMrfBacklogTracker {
     let mut tracker = DurableMrfBacklogTracker {
         available: true,
@@ -716,10 +712,6 @@ pub struct ReplicationPool<S: ReplicationStorage> {
 
     // MRF worker lifecycle
     mrf_worker_cancellations: Mutex<Vec<CancellationToken>>,
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     mrf_stop_tx: Sender<()>,
 
     // Worker size tracking
@@ -948,10 +940,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Resizes worker priority and counts
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     pub async fn resize_worker_priority(
         &self,
         pri: ReplicationPriority,
@@ -1192,10 +1180,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Queues an MRF save operation
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
         let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await;
     }
@@ -1667,10 +1651,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling regular replication operations
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     async fn add_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1684,10 +1664,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling large object replication operations
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     async fn add_large_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1702,10 +1678,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling MRF (Most Recent Failures) operations
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     async fn add_mrf_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1719,10 +1691,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Delete resync metadata from replication resync state in memory
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     pub async fn delete_resync_metadata(&self, bucket: &str) {
         let mut status_map = self.resyncer.status_map.write().await;
         status_map.remove(bucket);

crates/ecstore/src/bucket/replication/replication_resync_boundary.rs

@@ -21,31 +21,11 @@ pub(crate) use rustfs_replication::{
     should_count_head_proxy_failure,
 };
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT;
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION;
 pub(crate) const RESYNC_FILE_MAX_BYTES: usize = rustfs_replication::RESYNC_FILE_MAX_BYTES;
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const WIRE_ZERO_TIME_UNIX: i64 = rustfs_replication::resync::WIRE_ZERO_TIME_UNIX;
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const MRF_META_FORMAT: u16 = rustfs_replication::mrf::MRF_META_FORMAT;
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const MRF_META_VERSION: u16 = rustfs_replication::mrf::MRF_META_VERSION;
 
 fn map_replication_error(err: rustfs_replication::Error) -> Error {

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -122,10 +122,6 @@ const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[
     "tcp connect error",
 ];
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-)]
 const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60);
 
 static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new();
@@ -332,10 +328,6 @@ fn bounded_resync_max_jobs(value: usize) -> usize {
 #[derive(Debug)]
 pub struct ReplicationResyncer {
     pub status_map: Arc<RwLock<HashMap<String, BucketReplicationResyncStatus>>>,
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     pub worker_size: usize,
     pub(crate) cancel_tokens: Arc<RwLock<HashMap<ResyncCancelKey, CancellationToken>>>,
     resync_admission: Arc<Semaphore>,
@@ -552,10 +544,6 @@ impl ReplicationResyncer {
             .is_some_and(|status| status.failed_count > 0)
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     pub async fn persist_to_disk<S>(&self, cancel_token: CancellationToken, api: Arc<S>)
     where
         S: ReplicationObjectIO,

crates/ecstore/src/bucket/replication/replication_state.rs

@@ -340,10 +340,6 @@ impl ReplicationStats {
     }
 
     /// Site replication update replica statistics
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     fn sr_update_replica_stat(&self, size: i64) {
         self.sr_stats.replica_size.fetch_add(size, Ordering::Relaxed);
         self.sr_stats.replica_count.fetch_add(1, Ordering::Relaxed);
@@ -708,12 +704,6 @@ impl ReplicationStats {
         } else {
             BucketReplicationStats::new()
         };
-        // Stamp the serializable failure windows from the live samples: the
-        // samples themselves do not cross the peer-RPC wire, so this snapshot
-        // is what cluster aggregation and the metrics endpoints see.
-        for stat in replication_stats.stats.values_mut() {
-            stat.fail_stats.refresh_windows();
-        }
         let uptime = if cache.contains_key(bucket) {
             SystemTime::now()
                 .duration_since(SystemTime::UNIX_EPOCH)

crates/ecstore/src/bucket/replication/replication_stats_boundary.rs

@@ -15,9 +15,7 @@
 #[cfg(test)]
 pub(crate) use rustfs_replication::FailStats;
 pub(crate) use rustfs_replication::{
-    ActiveWorkerStat, ProxyMetric, ProxyStatsCache, QueueCache, ReplicationMetricScope, SRMetricsSummary,
+    ActiveWorkerStat, BucketReplicationStat, InQueueMetric, ProxyMetric, ProxyStatsCache, QueueCache, ReplicationMetricScope,
+    SRMetricsSummary, XferStats,
 };
-// Public so the admin wire DTOs (rustfs/src/admin/replication_metrics_wire.rs)
+pub use rustfs_replication::{BucketReplicationStats, BucketStats};
-// can project the internal stats onto the minio-go response shapes through
-// the storage_api facade chain.
-pub use rustfs_replication::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};

crates/ecstore/src/bucket/replication/replication_target_boundary.rs

@@ -27,10 +27,8 @@ use rustfs_utils::http::{
     AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID,
     AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE,
     HeaderExt as _, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
-    SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
+    SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map,
-    SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP,
+    is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key, ssec_replication_transport_header,
-    get_str, insert_header_map, is_internal_key, is_object_encryption_marker, is_replication_stripped_encryption_key,
-    ssec_replication_transport_header,
 };
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
@@ -121,27 +119,6 @@ fn classify_replication_source_encryption(metadata: &HashMap<String, String>) ->
     }
 }
 
-fn is_legacy_source_replication_timestamp_key(key: &str) -> bool {
-    fn has_prefix_and_suffix(key: &str, prefix: &str, suffix: &str) -> bool {
-        let key = key.as_bytes();
-        key.len() == prefix.len() + suffix.len()
-            && key[..prefix.len()].eq_ignore_ascii_case(prefix.as_bytes())
-            && key[prefix.len()..].eq_ignore_ascii_case(suffix.as_bytes())
-    }
-
-    [
-        SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP,
-        SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
-        SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
-    ]
-    .iter()
-    .any(|suffix| {
-        ["x-rustfs-", "x-minio-"]
-            .iter()
-            .any(|prefix| has_prefix_and_suffix(key, prefix, suffix))
-    })
-}
-
 pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap<String, String>) -> bool {
     rustfs_replication::is_ssec_encrypted(user_defined)
 }
@@ -199,11 +176,6 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
             continue;
         }
 
-        if is_legacy_source_replication_timestamp_key(key) {
-            meta.insert(format!("x-amz-meta-{key}"), value.to_string());
-            continue;
-        }
-
         if is_internal_key(key) || is_standard_header(key) {
             continue;
         }
@@ -287,23 +259,15 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
 
         if !tags.is_empty() {
             put_options.user_tags = tags;
+            put_options.internal.tagging_timestamp =
+                if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) {
+                    OffsetDateTime::parse(&timestamp, &Rfc3339)
+                        .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
+                } else {
+                    object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
+                };
         }
     }
-    // Load the stored tagging timestamp independently of whether any tags
-    // remain: DeleteObjectTagging leaves the object tagless but stamps this
-    // key, and the deletion's LWW timestamp must still reach the replica.
-    // With no stored key, fall back to mod_time only while tags exist
-    // (MinIO parity); a tagless object without the key was never tagged and
-    // keeps the epoch default (no header).
-    put_options.internal.tagging_timestamp = if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP)
-    {
-        OffsetDateTime::parse(&timestamp, &Rfc3339)
-            .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
-    } else if !put_options.user_tags.is_empty() {
-        object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-    } else {
-        OffsetDateTime::UNIX_EPOCH
-    };
 
     let metadata = &*object_info.user_defined;
 
@@ -319,15 +283,13 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
         put_options.cache_control = cache_control.to_string();
     }
 
-    if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE).filter(|mode| !mode.is_empty()) {
+    if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE) {
         put_options.mode = Some(ObjectLockRetentionMode::from(mode.to_uppercase().as_str()));
     }
 
     if let Some(retain_until_date) = metadata.lookup(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) {
-        if !retain_until_date.is_empty() {
+        put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339)
-            put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339)
+            .map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?;
-                .map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?;
-        }
         put_options.internal.retention_timestamp =
             if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP) {
                 OffsetDateTime::parse(&timestamp, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH)
@@ -732,110 +694,6 @@ mod tests {
         assert!(options.internal.replication_request);
     }
 
-    /// DeleteObjectTagging leaves the object tagless but stamps the
-    /// tagging-timestamp internal key; the deletion's LWW timestamp must
-    /// still be loaded (and therefore sent) so the replica can order the
-    /// deletion against concurrent tag edits.
-    #[test]
-    fn replication_put_options_carry_tagging_timestamp_after_tag_deletion() {
-        let mut metadata = std::collections::HashMap::new();
-        rustfs_utils::http::insert_str(&mut metadata, SUFFIX_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
-
-        let object_info = ObjectInfo {
-            user_defined: Arc::new(metadata),
-            user_tags: Arc::new(String::new()),
-            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
-            version_id: Some(Uuid::nil()),
-            ..Default::default()
-        };
-
-        let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
-
-        assert!(options.user_tags.is_empty());
-        assert_eq!(
-            options.internal.tagging_timestamp,
-            OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp"),
-            "the stored tagging timestamp must load independently of remaining tags"
-        );
-
-        // A tagless object without the stored key was never tagged: the epoch
-        // default keeps the header unsent.
-        let untagged = ObjectInfo {
-            user_tags: Arc::new(String::new()),
-            mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
-            version_id: Some(Uuid::nil()),
-            ..Default::default()
-        };
-        let (options, _) = replication_put_object_options("", &untagged).expect("build put options");
-        assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
-    }
-
-    #[test]
-    fn replication_put_options_do_not_promote_legacy_user_timestamp_metadata() {
-        let legacy_keys = [
-            "x-rustfs-source-replication-tagging-timestamp",
-            "x-rustfs-source-replication-retention-timestamp",
-            "x-rustfs-source-replication-legalhold-timestamp",
-            "x-minio-source-replication-tagging-timestamp",
-            "x-minio-source-replication-retention-timestamp",
-            "x-minio-source-replication-legalhold-timestamp",
-        ];
-        let object_info = ObjectInfo {
-            user_defined: Arc::new(
-                legacy_keys
-                    .iter()
-                    .map(|key| (key.to_string(), "2099-01-02T03:04:05Z".to_string()))
-                    .collect(),
-            ),
-            ..Default::default()
-        };
-
-        let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
-
-        for legacy_key in legacy_keys {
-            assert!(!options.user_metadata.contains_key(legacy_key));
-            assert_eq!(
-                options
-                    .user_metadata
-                    .get(&format!("x-amz-meta-{legacy_key}"))
-                    .map(String::as_str),
-                Some("2099-01-02T03:04:05Z")
-            );
-        }
-        assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
-        assert_eq!(options.internal.retention_timestamp, OffsetDateTime::UNIX_EPOCH);
-        assert_eq!(options.internal.legalhold_timestamp, OffsetDateTime::UNIX_EPOCH);
-    }
-
-    #[test]
-    fn replication_put_options_carry_retention_timestamp_after_clear() {
-        let mut metadata = HashMap::from([
-            (AMZ_OBJECT_LOCK_MODE.to_string(), String::new()),
-            (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.to_string(), String::new()),
-        ]);
-        rustfs_utils::http::insert_str(&mut metadata, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
-        let object_info = ObjectInfo {
-            user_defined: Arc::new(metadata),
-            ..Default::default()
-        };
-
-        let (options, _) = replication_put_object_options("", &object_info).expect("retention clear must replicate");
-
-        assert!(options.mode.is_none());
-        assert_eq!(options.retain_until_date, OffsetDateTime::UNIX_EPOCH);
-        assert_eq!(
-            options.internal.retention_timestamp,
-            OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp")
-        );
-        let headers = options.header();
-        assert!(!headers.contains_key(AMZ_OBJECT_LOCK_MODE));
-        assert!(!headers.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE));
-        assert_eq!(
-            rustfs_utils::http::get_header(&headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP).as_deref(),
-            Some("2026-01-02T03:04:05Z")
-        );
-    }
-
     #[test]
     fn replication_put_options_strip_encryption_metadata_from_plaintext_objects() {
         use rustfs_utils::http::object_encryption_keys::{INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER, SSEC_ORIGINAL_SIZE_HEADER};

crates/ecstore/src/bucket/target/arn.rs

@@ -40,14 +40,7 @@ impl ARN {
 
 impl Display for ARN {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        // The `minio` partition is deliberate: madmin-go's ParseARN
+        write!(f, "arn:rustfs:{}:{}:{}:{}", self.arn_type, self.region, self.id, self.bucket)
-        // hard-rejects any other partition, so native mc/madmin tooling can
-        // only decode remote-target ARNs minted in this form (backlog#1675
-        // P1-7). Legacy `arn:rustfs:` ARNs persisted by older releases stay
-        // readable via the FromStr whitelist below; runtime matching between
-        // targets and replication rules is by full-string equality, so mixed
-        // partitions coexist safely.
-        write!(f, "arn:minio:{}:{}:{}:{}", self.arn_type, self.region, self.id, self.bucket)
     }
 }
 
@@ -55,12 +48,7 @@ impl FromStr for ARN {
     type Err = std::io::Error;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        // Partition whitelist, not just an `arn:` check: `BucketTargetType::
+        if !s.starts_with("arn:rustfs:") {
-        // from_str(...).unwrap_or_default()` below never fails, so this is
-        // the only structural gate rejecting foreign ARNs. `arn:rustfs:` is
-        // the legacy partition and must stay accepted forever (persisted
-        // bucket-targets.json / replication configs from older releases).
-        if !s.starts_with("arn:minio:") && !s.starts_with("arn:rustfs:") {
             return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "Invalid ARN format"));
         }
 
@@ -113,50 +101,14 @@ mod tests {
     }
 
     /// RustFS commonly generates ARNs with an empty region:
-    /// `arn:minio:replication::<deployment_id>:<bucket>`.
+    /// `arn:rustfs:replication::<deployment_id>:<bucket>`.
     #[test]
     fn from_str_handles_empty_region_segment() {
-        let parsed = ARN::from_str("arn:minio:replication::depl-123:bucket-a").expect("valid ARN must parse");
+        let parsed = ARN::from_str("arn:rustfs:replication::depl-123:bucket-a").expect("valid ARN must parse");
 
         assert_eq!(parsed.arn_type, BucketTargetType::ReplicationService);
         assert_eq!(parsed.region, "", "region segment is empty in this form");
         assert_eq!(parsed.id, "depl-123");
         assert_eq!(parsed.bucket, "bucket-a");
     }
-
-    /// madmin-go's `ParseARN` hard-rejects anything that does not start with
-    /// `arn:minio:`, so generated ARNs must use the `minio` partition or the
-    /// native mc/madmin tooling cannot decode remote-target listings.
-    #[test]
-    fn display_emits_minio_partition() {
-        let arn = ARN::new(
-            BucketTargetType::ReplicationService,
-            "depl-123".to_string(),
-            String::new(),
-            "bucket-a".to_string(),
-        );
-
-        assert_eq!(arn.to_string(), "arn:minio:replication::depl-123:bucket-a");
-    }
-
-    /// Persisted bucket-targets.json files from older RustFS releases carry
-    /// `arn:rustfs:` ARNs; the legacy partition must stay parseable forever.
-    #[test]
-    fn from_str_accepts_legacy_rustfs_partition() {
-        let parsed = ARN::from_str("arn:rustfs:replication:us-east-1:depl-123:bucket-a").expect("legacy ARN must parse");
-
-        assert_eq!(parsed.arn_type, BucketTargetType::ReplicationService);
-        assert_eq!(parsed.region, "us-east-1");
-        assert_eq!(parsed.id, "depl-123");
-        assert_eq!(parsed.bucket, "bucket-a");
-    }
-
-    /// The partition whitelist is the only structural gate: `BucketTargetType::
-    /// from_str(...).unwrap_or_default()` never fails, so any 6-segment string
-    /// would otherwise parse as `type=None`.
-    #[test]
-    fn from_str_rejects_unknown_partition() {
-        assert!(ARN::from_str("arn:aws:replication::depl-123:bucket-a").is_err());
-        assert!(ARN::from_str("not-an-arn").is_err());
-    }
 }

crates/ecstore/src/bucket/target/bucket_target.rs

@@ -59,10 +59,6 @@ impl fmt::Debug for Credentials {
 }
 
 #[derive(Debug, Deserialize, Serialize, Default, Clone)]
-#[allow(
-    dead_code,
-    reason = "MinIO-parity bucket-target service discriminator with no caller in this port (backlog#1823)"
-)]
 pub enum ServiceType {
     #[default]
     Replication,

crates/ecstore/src/bucket/utils.rs

@@ -73,6 +73,23 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> {
     check_bucket_name_common(bucket_name, true)
 }
 
+pub fn check_valid_object_name_prefix(object_name: &str) -> Result<()> {
+    if object_name.len() > 1024 {
+        return Err(Error::other("Object name cannot be longer than 1024 characters"));
+    }
+    if !object_name.is_ascii() {
+        return Err(Error::other("Object name with non-UTF-8 strings are not supported"));
+    }
+    Ok(())
+}
+
+pub fn check_valid_object_name(object_name: &str) -> Result<()> {
+    if object_name.trim().is_empty() {
+        return Err(Error::other("Object name cannot be empty"));
+    }
+    check_valid_object_name_prefix(object_name)
+}
+
 pub fn deserialize<T>(input: &[u8]) -> xml::DeResult<T>
 where
     T: for<'xml> xml::Deserialize<'xml>,
@@ -83,10 +100,6 @@ where
     Ok(ans)
 }
 
-#[allow(
-    dead_code,
-    reason = "xml serialize helper with no caller in this port; the live sibling is deserialize (backlog#1823)"
-)]
 pub fn serialize_content<T: xml::SerializeContent>(val: &T) -> xml::SerResult<String> {
     let mut buf = Vec::with_capacity(256);
     {
@@ -173,27 +186,15 @@ pub fn is_valid_object_name(object: &str) -> bool {
 /// Client-facing reason attached to rejections of object keys that Win32/NTFS
 /// cannot represent as file paths (issue #3299). Deployments on Linux/macOS
 /// accept the full S3 key character set.
-#[allow(
-    dead_code,
-    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
-)]
 pub const WINDOWS_RESERVED_CHARACTERS_REASON: &str =
     "object key contains characters unsupported on Windows hosts (one of ':', '*', '?', '\"', '|', '<', '>')";
 
 /// Client-facing reason for path segments Windows can store but not address
 /// afterwards (issue #3449): trailing dot/space or reserved DOS device names.
-#[allow(
-    dead_code,
-    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
-)]
 pub const WINDOWS_RESERVED_SEGMENT_REASON: &str = "object key contains a path segment unsupported on Windows hosts (trailing dot or space, or a reserved device name such as NUL/CON/COM1)";
 
 /// Reserved DOS device names that shadow regular files on Windows, even when
 /// an extension is appended (e.g. `NUL.txt` resolves to the `NUL` device).
-#[allow(
-    dead_code,
-    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
-)]
 const WINDOWS_RESERVED_NAMES: &[&str] = &[
     "CON", "PRN", "AUX", "NUL", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9", "LPT1", "LPT2", "LPT3",
     "LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9",
@@ -203,10 +204,6 @@ const WINDOWS_RESERVED_NAMES: &[&str] = &[
 /// the Win32 API cannot address afterwards (issue #3449): segments ending in a
 /// dot or a space, and reserved DOS device names — bare or with an extension
 /// (`NUL.txt`), matching classic Win32 path resolution semantics.
-#[allow(
-    dead_code,
-    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
-)]
 pub fn object_name_has_windows_incompatible_segment(object: &str) -> bool {
     object.split(['/', '\\']).any(|segment| {
         if segment.ends_with('.') || segment.ends_with(' ') {

crates/ecstore/src/bucket/versioning_sys.rs

@@ -90,10 +90,6 @@ impl BucketVersioningSys {
     /// caller's own instance context so a second in-process store never
     /// answers with the first instance's versioning state; falls back to the
     /// ambient system when the instance cell is not initialized.
-    #[allow(
-        dead_code,
-        reason = "instance-scoped seam (backlog#1052) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<VersioningConfiguration> {
         if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) {
             return Ok(VersioningConfiguration::default());

crates/ecstore/src/cache_value/metacache_set.rs

@@ -15,7 +15,6 @@
 use crate::disk::disk_store::{get_drive_walkdir_peek_timeout, get_drive_walkdir_stall_timeout};
 use crate::disk::error::DiskError;
 use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions};
-use futures::future::join_all;
 use metrics::counter;
 use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetacacheReader, is_io_eof};
 use std::{
@@ -656,7 +655,6 @@ async fn list_path_raw_inner(
             errs.push(None);
         }
         let mut pending_entries: Vec<Option<MetaCacheEntry>> = vec![None; readers.len()];
-        let mut peek_outcomes: Vec<Option<PeekOutcome>> = std::iter::repeat_with(|| None).take(readers.len()).collect();
 
         loop {
             let mut current = MetaCacheEntry::default();
@@ -678,21 +676,6 @@ async fn list_path_raw_inner(
             let mut has_err = 0;
             let mut agree = 0;
 
-            // Start every missing head read in the same round so one stalled
-            // disk cannot multiply the wait budget by the erasure-set width.
-            // Outcomes are still consumed below in stable disk-index order.
-            let concurrent_peeks = readers.iter_mut().enumerate().filter_map(|(i, reader)| {
-                if errs[i].is_some() || pending_entries[i].is_some() {
-                    return None;
-                }
-
-                let cancel = &revjob_rx;
-                Some(async move { (i, peek_with_timeout(cancel, reader, peek_timeout).await) })
-            });
-            for (i, outcome) in join_all(concurrent_peeks).await {
-                peek_outcomes[i] = Some(outcome);
-            }
-
             for (i, r) in readers.iter_mut().enumerate() {
                 if errs[i].is_some() {
                     has_err += 1;
@@ -702,10 +685,7 @@ async fn list_path_raw_inner(
                 let entry = if let Some(entry) = pending_entries[i].take() {
                     entry
                 } else {
-                    let Some(outcome) = peek_outcomes[i].take() else {
+                    match peek_with_timeout(&revjob_rx, r, peek_timeout).await {
-                        return Err(DiskError::Unexpected);
-                    };
-                    match outcome {
                         PeekOutcome::Ready(res) => {
                             if let Some(entry) = res {
                                 // info!("read entry disk: {}, name: {}", i, entry.name);
@@ -1315,36 +1295,6 @@ mod tests {
         assert_eq!(err, DiskError::Timeout);
     }
 
-    #[tokio::test(start_paused = true)]
-    async fn list_path_raw_bounds_multiple_stalled_readers_by_one_peek_deadline() {
-        let peek_timeout = Duration::from_millis(20);
-        let started = tokio::time::Instant::now();
-        let err = list_path_raw(
-            CancellationToken::new(),
-            ListPathRawOptions {
-                disks: vec![None, None, None, None],
-                min_disks: 1,
-                test_reader_behaviors: vec![
-                    TestReaderBehavior::Stall,
-                    TestReaderBehavior::Stall,
-                    TestReaderBehavior::Stall,
-                    TestReaderBehavior::Stall,
-                ],
-                peek_timeout: Some(peek_timeout),
-                ..Default::default()
-            },
-        )
-        .await
-        .expect_err("all stalled readers should fail the listing");
-
-        assert_eq!(err, DiskError::Timeout);
-        assert_eq!(
-            started.elapsed(),
-            peek_timeout,
-            "reader deadlines must overlap instead of accumulating once per disk"
-        );
-    }
-
     #[tokio::test]
     async fn list_path_raw_waits_past_producer_stall_for_slow_progressing_reader() {
         let entry = MetaCacheEntry {

crates/ecstore/src/client/api_error_response.rs

@@ -229,6 +229,17 @@ pub fn http_resp_to_error_response(
     err_resp
 }
 
+pub fn err_transfer_acceleration_bucket(bucket_name: &str) -> ErrorResponse {
+    ErrorResponse {
+        status_code: StatusCode::BAD_REQUEST,
+        code: S3ErrorCode::InvalidArgument,
+        message: "The name of the bucket used for Transfer Acceleration must be DNS-compliant and must not contain periods ‘.’."
+            .to_string(),
+        bucket_name: bucket_name.to_string(),
+        ..Default::default()
+    }
+}
+
 pub fn err_entity_too_large(total_size: i64, max_object_size: i64, bucket_name: &str, object_name: &str) -> ErrorResponse {
     let msg = format!(
         "Your proposed upload size ‘{}’ exceeds the maximum allowed object size ‘{}’ for single PUT operation.",
@@ -284,6 +295,16 @@ pub fn err_invalid_argument(message: &str) -> ErrorResponse {
     }
 }
 
+pub fn err_api_not_supported(message: &str) -> ErrorResponse {
+    ErrorResponse {
+        status_code: StatusCode::NOT_IMPLEMENTED,
+        code: S3ErrorCode::Custom("APINotSupported".into()),
+        message: message.to_string(),
+        request_id: "rustfs".to_string(),
+        ..Default::default()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/ecstore/src/client/api_get_object.rs

@@ -135,10 +135,6 @@ impl Object {
         Self { ..Default::default() }
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity reader surface with no caller in this port (backlog#1823)"
-    )]
     fn do_get_request(&self, request: &GetRequest) -> Result<GetResponse, std::io::Error> {
         let _ = request.did_offset_change;
         let _ = request.offset;
@@ -154,20 +150,12 @@ impl Object {
         ))
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
-    )]
     fn set_offset(&mut self, bytes_read: i64) -> Result<(), std::io::Error> {
         self.curr_offset += bytes_read;
 
         Ok(())
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
-    )]
     fn read(&mut self, b: &[u8]) -> Result<i64, std::io::Error> {
         let mut read_req = GetRequest {
             is_read_op: true,
@@ -192,10 +180,6 @@ impl Object {
         Ok(response.size)
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
-    )]
     fn stat(&self) -> Result<ObjectInfo, std::io::Error> {
         if !self.is_started || !self.object_info_set {
             let _ = self.do_get_request(&GetRequest {
@@ -208,10 +192,6 @@ impl Object {
         Ok(self.object_info.clone())
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
-    )]
     fn read_at(&mut self, b: &[u8], offset: i64) -> Result<i64, std::io::Error> {
         self.curr_offset = offset;
 
@@ -239,10 +219,6 @@ impl Object {
         Ok(response.size)
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
-    )]
     fn seek(&mut self, offset: i64, whence: i64) -> Result<i64, std::io::Error> {
         if !self.is_started || !self.object_info_set {
             let seek_req = GetRequest {
@@ -277,10 +253,6 @@ impl Object {
         Ok(self.curr_offset)
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity Object reader method with no caller in this port (backlog#1823)"
-    )]
     fn close(&mut self) -> Result<(), std::io::Error> {
         self.is_closed = true;
         Ok(())

crates/ecstore/src/client/api_put_object.rs

@@ -37,7 +37,7 @@ use crate::client::{
     api_put_object_common::optimal_part_info,
     api_put_object_multipart::UploadPartParams,
     api_s3_datatypes::{CompleteMultipartUpload, CompletePart, ObjectPart},
-    constants::{ISO8601_DATEFORMAT, MAX_MULTIPART_PUT_OBJECT_SIZE, MIN_PART_SIZE},
+    constants::{ISO8601_DATEFORMAT, MAX_MULTIPART_PUT_OBJECT_SIZE, MIN_PART_SIZE, TOTAL_WORKERS},
     credentials::SignatureType,
     transition_api::{ReaderImpl, TransitionClient, UploadInfo},
     utils::{is_amz_header, is_minio_header, is_rustfs_header, is_standard_header, is_storageclass_header},

crates/ecstore/src/client/api_put_object_common.rs

@@ -30,6 +30,10 @@ pub fn is_object(reader: &ReaderImpl) -> bool {
     matches!(reader, ReaderImpl::ObjectBody(_))
 }
 
+pub fn is_read_at(reader: ReaderImpl) -> bool {
+    matches!(reader, ReaderImpl::ObjectBody(_))
+}
+
 pub fn optimal_part_info(object_size: i64, configured_part_size: u64) -> Result<(i64, i64, i64), std::io::Error> {
     let unknown_size;
     let mut object_size = object_size;

crates/ecstore/src/client/api_put_object_streaming.rs

@@ -81,6 +81,18 @@ async fn read_multipart_part(reader: &mut ReaderImpl, want: usize) -> Result<Vec
     }
 }
 
+pub struct UploadedPartRes {
+    pub error: std::io::Error,
+    pub part_num: i64,
+    pub size: i64,
+    pub part: ObjectPart,
+}
+
+pub struct UploadPartReq {
+    pub part_num: i64,
+    pub part: ObjectPart,
+}
+
 impl TransitionClient {
     pub async fn put_object_multipart_stream(
         self: Arc<Self>,

crates/ecstore/src/client/api_s3_datatypes.rs

@@ -29,6 +29,10 @@ use crate::client::utils::base64_decode;
 
 use super::transition_api;
 
+pub struct ListAllMyBucketsResult {
+    pub owner: Owner,
+}
+
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct CommonPrefix {
     pub prefix: String,
@@ -85,10 +89,6 @@ pub struct ListVersionsResult {
     pub next_version_id_marker: String,
 }
 
-#[allow(
-    dead_code,
-    reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
-)]
 pub struct ListBucketResult {
     common_prefixes: Vec<CommonPrefix>,
     contents: Vec<transition_api::ObjectInfo>,
@@ -102,10 +102,6 @@ pub struct ListBucketResult {
     prefix: String,
 }
 
-#[allow(
-    dead_code,
-    reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
-)]
 pub struct ListMultipartUploadsResult {
     bucket: String,
     key_marker: String,
@@ -121,15 +117,16 @@ pub struct ListMultipartUploadsResult {
     common_prefixes: Vec<CommonPrefix>,
 }
 
-#[allow(
-    dead_code,
-    reason = "fields of a MinIO-parity list result that this port builds but never reads back (backlog#1823)"
-)]
 pub struct Initiator {
     id: String,
     display_name: String,
 }
 
+pub struct CopyObjectResult {
+    pub etag: String,
+    pub last_modified: OffsetDateTime,
+}
+
 #[derive(Debug, Clone)]
 pub struct ObjectPart {
     pub etag: String,
@@ -263,7 +260,6 @@ pub struct CompletePart {
 }
 
 impl CompletePart {
-    #[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
     fn checksum(&self, t: &ChecksumMode) -> String {
         match t {
             ChecksumMode::ChecksumCRC32C => {
@@ -288,6 +284,11 @@ impl CompletePart {
     }
 }
 
+pub struct CopyObjectPartResult {
+    pub etag: String,
+    pub last_modified: OffsetDateTime,
+}
+
 #[derive(Debug, Default, serde::Serialize)]
 #[serde(rename = "CompleteMultipartUpload")]
 pub struct CompleteMultipartUpload {
@@ -356,10 +357,10 @@ impl CompleteMultipartUpload {
     }
 }
 
-#[allow(
+pub struct CreateBucketConfiguration {
-    dead_code,
+    pub location: String,
-    reason = "live via quick_xml::de::from_str in bucket_cache.rs; serde deserialization is not a construction (backlog#1823)"
+}
-)]
+
 #[derive(serde::Serialize)]
 pub struct DeleteObject {
     //api has
@@ -367,6 +368,21 @@ pub struct DeleteObject {
     pub version_id: String,
 }
 
+pub struct DeletedObject {
+    //s3s has
+    pub key: String,
+    pub version_id: String,
+    pub deletemarker: bool,
+    pub deletemarker_version_id: String,
+}
+
+pub struct NonDeletedObject {
+    pub key: String,
+    pub code: String,
+    pub message: String,
+    pub version_id: String,
+}
+
 #[derive(serde::Serialize)]
 pub struct DeleteMultiObjects {
     pub quiet: bool,
@@ -386,7 +402,6 @@ impl DeleteMultiObjects {
         Ok(buf)
     }
 
-    #[allow(dead_code, reason = "MinIO-parity XML helper with no caller in this port (backlog#1823)")]
     pub fn unmarshal(buf: &[u8]) -> Result<Self, std::io::Error> {
         #[derive(Debug, Deserialize)]
         struct WireDeleteObject {
@@ -421,3 +436,8 @@ impl DeleteMultiObjects {
         })
     }
 }
+
+pub struct DeleteMultiObjectsResult {
+    pub deleted_objects: Vec<DeletedObject>,
+    pub undeleted_objects: Vec<NonDeletedObject>,
+}

crates/ecstore/src/client/checksum.rs

@@ -365,10 +365,6 @@ mod tests {
 pub struct Checksum {
     checksum_type: ChecksumMode,
     r: Vec<u8>,
-    #[allow(
-        dead_code,
-        reason = "checksum bookkeeping field kept beside the value it guards (backlog#1823)"
-    )]
     computed: bool,
 }
 

crates/ecstore/src/client/constants.rs

@@ -32,5 +32,8 @@ pub const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
 pub const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
 pub const UNSIGNED_PAYLOAD_TRAILER: &str = "STREAMING-UNSIGNED-PAYLOAD-TRAILER";
 
+pub const TOTAL_WORKERS: i64 = 4;
+
+pub const SIGN_V4_ALGORITHM: &str = "AWS4-HMAC-SHA256";
 pub const ISO8601_DATEFORMAT: &[FormatItem<'_>] =
     format_description!("[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond]Z");

crates/ecstore/src/client/credentials.rs

@@ -67,10 +67,6 @@ impl<P: Provider + Default> Credentials<P> {
         Ok(self.creds.clone())
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity credential surface with no caller in this port (backlog#1823)"
-    )]
     fn expire(&mut self) {
         self.force_refresh = true;
     }
@@ -137,10 +133,6 @@ impl Provider for Static {
 
 #[derive(Debug, Clone, Default)]
 pub struct STSError {
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity STS error detail that this port never reads back (backlog#1823)"
-    )]
     pub r#type: String,
     pub code: String,
     pub message: String,
@@ -149,10 +141,6 @@ pub struct STSError {
 #[derive(Debug, Clone, thiserror::Error)]
 pub struct ErrorResponse {
     pub sts_error: STSError,
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity STS error detail that this port never reads back (backlog#1823)"
-    )]
     pub request_id: String,
 }
 
@@ -170,3 +158,22 @@ impl ErrorResponse {
         return self.sts_error.message.clone();
     }
 }
+
+pub fn xml_decoder<T>(body: &[u8]) -> Result<T, Error>
+where
+    for<'de> T: Deserialize<'de>,
+{
+    match std::str::from_utf8(body) {
+        Ok(xml_body) => quick_xml::de::from_str::<T>(xml_body).map_err(|err| Error::new(ErrorKind::InvalidData, err.to_string())),
+        Err(err) => Err(Error::new(ErrorKind::InvalidData, err.to_string())),
+    }
+}
+
+pub fn xml_decode_and_body<T>(body_reader: &[u8]) -> Result<(Vec<u8>, T), std::io::Error>
+where
+    for<'de> T: Deserialize<'de>,
+{
+    let body = body_reader.to_vec();
+    let parsed = xml_decoder(&body)?;
+    Ok((body, parsed))
+}

crates/ecstore/src/client/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 // #730: S3 client compatibility models are kept while ECStore callers move to narrower facades.
+#![allow(dead_code)]
 
 pub mod admin_handler_utils;
 pub mod api_error_response;

crates/ecstore/src/client/object_api_utils.rs

@@ -77,6 +77,39 @@ fn part_number_to_rangespec(oi: ObjectInfo, part_number: usize) -> Option<HTTPRa
     })
 }
 
+fn get_compressed_offsets(oi: ObjectInfo, offset: i64) -> (i64, i64, i64, i64, u64) {
+    let mut skip_length: i64 = 0;
+    let mut cumulative_actual_size: i64 = 0;
+    let mut first_part_idx: i64 = 0;
+    let mut compressed_offset: i64 = 0;
+    let mut part_skip: i64 = 0;
+    let mut decrypt_skip: i64 = 0;
+    let mut seq_num: u64 = 0;
+    for (i, part) in oi.parts.iter().enumerate() {
+        cumulative_actual_size += part.actual_size as i64;
+        if cumulative_actual_size <= offset {
+            compressed_offset += part.size as i64;
+        } else {
+            first_part_idx = i as i64;
+            skip_length = cumulative_actual_size - part.actual_size as i64;
+            break;
+        }
+    }
+    skip_length = offset - skip_length;
+
+    let parts: &[ObjectPartInfo] = &oi.parts;
+    if skip_length > 0
+        && parts.len() > first_part_idx as usize
+        && parts[first_part_idx as usize].index.as_ref().is_some_and(|idx| idx.len() > 0)
+    {
+        let _ = part_skip;
+        let _ = decrypt_skip;
+        let _ = seq_num;
+    }
+
+    (compressed_offset, part_skip, first_part_idx, decrypt_skip, seq_num)
+}
+
 pub fn new_getobjectreader<'a>(
     rs: &Option<HTTPRangeSpec>,
     oi: &'a ObjectInfo,

crates/ecstore/src/client/provider_versions.rs

@@ -23,7 +23,6 @@ const X_OBS_VERSION_ID: &str = "x-obs-version-id";
 const MAX_REMOTE_VERSION_ID_LEN: usize = 1024;
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
-#[allow(dead_code, reason = "bucket versioning states kept as a complete vocabulary (backlog#1823)")]
 pub(crate) enum BucketVersioningState {
     Unknown,
     Disabled,
@@ -48,7 +47,6 @@ impl RemoteVersion {
         }
     }
 
-    #[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
     pub(crate) fn exact_request_id(&self) -> Result<Option<&str>, Error> {
         match self {
             Self::Unknown => Err(Error::new(

crates/ecstore/src/client/transition_api.rs

@@ -101,10 +101,6 @@ where
 
 const C_UNKNOWN: i32 = -1;
 const C_OFFLINE: i32 = 0;
-#[allow(
-    dead_code,
-    reason = "reachable only from the unused transition client methods below (backlog#1823)"
-)]
 const C_ONLINE: i32 = 1;
 
 fn invalid_utf8_header_error(scope: &str, header_name: &str) -> std::io::Error {
@@ -324,10 +320,6 @@ impl TransitionClient {
         Ok(client)
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client surface with no caller in this port (backlog#1823)"
-    )]
     fn endpoint_url(&self) -> Url {
         self.endpoint_url.clone()
     }
@@ -356,20 +348,12 @@ impl TransitionClient {
             .to_string())
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
-    )]
     fn trace_errors_only_off(&self) {
         if let Ok(mut trace_errors_only) = self.trace_errors_only.lock() {
             *trace_errors_only = false;
         }
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
-    )]
     fn trace_off(&self) {
         if let Ok(mut is_trace_enabled) = self.is_trace_enabled.lock() {
             *is_trace_enabled = false;
@@ -379,20 +363,12 @@ impl TransitionClient {
         }
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
-    )]
     fn set_s3_transfer_accelerate(&self, accelerate_endpoint: &str) {
         if let Ok(mut endpoint) = self.s3_accelerate_endpoint.lock() {
             *endpoint = accelerate_endpoint.to_string();
         }
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
-    )]
     fn set_s3_enable_dual_stack(&self, enabled: bool) {
         if let Ok(mut dual_stack) = self.s3_dual_stack_enabled.lock() {
             *dual_stack = enabled;
@@ -422,18 +398,10 @@ impl TransitionClient {
         (hash_algos, hash_sums)
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
-    )]
     fn is_online(&self) -> bool {
         !self.is_offline()
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
-    )]
     fn mark_offline(&self) {
         self.health_status
             .compare_exchange(C_ONLINE, C_OFFLINE, Ordering::SeqCst, Ordering::SeqCst);
@@ -443,18 +411,10 @@ impl TransitionClient {
         self.health_status.load(Ordering::SeqCst) == C_OFFLINE
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
-    )]
     fn health_check(hc_duration: Duration) {
         let _ = hc_duration;
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity transition client method with no caller in this port (backlog#1823)"
-    )]
     fn dump_http(&self, req: &Request<s3s::Body>, resp: &Response<Incoming>) -> Result<(), std::io::Error> {
         let mut resp_trace: Vec<u8>;
 
@@ -1142,7 +1102,6 @@ impl Default for ObjectInfo {
 }
 
 impl ObjectInfo {
-    #[allow(dead_code, reason = "MinIO-parity accessor with no caller in this port (backlog#1823)")]
     pub(crate) fn remote_version(
         &self,
         capabilities: ProviderVersionCapabilities,

crates/ecstore/src/client/utils.rs

@@ -48,6 +48,10 @@ lazy_static! {
     };
 }
 
+pub fn is_standard_query_value(qs_key: &str) -> bool {
+    SUPPORTED_QUERY_VALUES[qs_key]
+}
+
 pub fn is_storageclass_header(header_key: &str) -> bool {
     header_key.to_lowercase() == X_AMZ_STORAGE_CLASS.as_str().to_lowercase()
 }

crates/ecstore/src/diagnostics/get.rs

@@ -190,17 +190,6 @@ pub(crate) const GET_METADATA_CACHE_REASON_VERSION_SUSPENDED: &str = "version_su
 pub(crate) const GET_METADATA_CACHE_REASON_VERSIONED: &str = "versioned";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA: &str = "conflicting_metadata";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER: &str = "delete_marker";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY: &str = "data_read_inline_body_verify";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED: &str = "data_read_inline_deleted";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY: &str = "data_read_inline_geometry";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH: &str = "data_read_inline_identity_mismatch";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD: &str = "data_read_inline_missing_payload";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD: &str = "data_read_inline_missing_shard";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE: &str = "data_read_inline_not_inline";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE: &str = "data_read_inline_part_shape";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE: &str = "data_read_inline_remote";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE: &str = "data_read_inline_size";
-pub(crate) const GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED: &str = "data_read_inline_transformed";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_ERROR: &str = "error";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM: &str = "insufficient_quorum";
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_NOT_FOUND: &str = "not_found";
@@ -562,32 +551,6 @@ mod tests {
         assert_eq!(GET_METADATA_CACHE_REASON_VERSIONED, "versioned");
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, "conflicting_metadata");
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, "delete_marker");
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
-            "data_read_inline_body_verify"
-        );
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, "data_read_inline_deleted");
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY, "data_read_inline_geometry");
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
-            "data_read_inline_identity_mismatch"
-        );
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
-            "data_read_inline_missing_payload"
-        );
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD,
-            "data_read_inline_missing_shard"
-        );
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE, "data_read_inline_not_inline");
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, "data_read_inline_part_shape");
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE, "data_read_inline_remote");
-        assert_eq!(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, "data_read_inline_size");
-        assert_eq!(
-            GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
-            "data_read_inline_transformed"
-        );
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_ERROR, "error");
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, "insufficient_quorum");
         assert_eq!(GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, "not_found");

crates/ecstore/src/disk/disk_store.rs

@@ -637,23 +637,14 @@ impl Default for DiskOperationMetrics {
 }
 
 impl DiskOperationMetrics {
-    #[allow(
-        dead_code,
-        reason = "internal metrics recorder reached only from record() below (backlog#1823)"
-    )]
     fn record_call(&mut self) {
         self.lifetime_calls.fetch_add(1, Ordering::Relaxed);
     }
 
-    #[allow(
-        dead_code,
-        reason = "internal metrics recorder reached only from record() below (backlog#1823)"
-    )]
     fn record_latency(&mut self, now_sec: u64, elapsed: Duration) {
         self.record_latency_atomic(now_sec, elapsed);
     }
 
-    #[allow(dead_code, reason = "metrics roll-up with no caller in this port (backlog#1823)")]
     fn record(&mut self, now_sec: u64, elapsed: Duration) {
         self.record_call();
         self.record_latency(now_sec, elapsed);
@@ -779,7 +770,6 @@ impl DiskHealthTracker {
     }
 
     /// Set disk as faulty
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn set_faulty(&self) {
         self.status.store(DISK_HEALTH_FAULTY, Ordering::Release);
     }
@@ -860,7 +850,6 @@ impl DiskHealthTracker {
         became_offline
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn mark_offline(&self, endpoint: &Endpoint, reason: &'static str) -> bool {
         let current = self.runtime_state();
         if current == RuntimeDriveHealthState::Offline {
@@ -991,13 +980,11 @@ impl DiskHealthTracker {
     }
 
     /// Get waiting operations count
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn waiting_count(&self) -> u32 {
         self.waiting.load(Ordering::Relaxed)
     }
 
     /// Get last success timestamp
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn last_success(&self) -> i64 {
         self.last_success.load(Ordering::Acquire)
     }
@@ -1039,6 +1026,21 @@ impl Default for DiskHealthTracker {
     }
 }
 
+/// Health check context key for tracking disk operations
+#[derive(Debug, Clone)]
+struct HealthDiskCtxKey;
+
+#[derive(Debug)]
+struct HealthDiskCtxValue {
+    last_success: Arc<AtomicI64>,
+}
+
+impl HealthDiskCtxValue {
+    fn log_success(&self) {
+        self.last_success.store(current_unix_nanos(), Ordering::Relaxed);
+    }
+}
+
 /// LocalDiskWrapper wraps a DiskStore with health tracking capabilities.
 /// This is similar to Go's xlStorageDiskIDCheck.
 #[derive(Debug, Clone)]
@@ -1070,6 +1072,10 @@ impl LocalDiskWrapper {
         )
     }
 
+    pub(crate) fn new_with_health(disk: Arc<LocalDisk>, health_check: bool, health: Arc<DiskHealthTracker>) -> Self {
+        Self::new_with_health_and_metrics(disk, health_check, health, Arc::new(DiskHealthMetricEpoch::default()))
+    }
+
     pub(crate) fn new_with_reconnect_state(
         disk: Arc<LocalDisk>,
         health_check: bool,
@@ -1432,6 +1438,20 @@ impl LocalDiskWrapper {
         }
     }
 
+    async fn check_id(&self, want_id: Option<Uuid>) -> Result<()> {
+        if want_id.is_none() {
+            return Ok(());
+        }
+
+        let stored_disk_id = self.disk.get_disk_id().await?;
+
+        if stored_disk_id != want_id {
+            return Err(Error::other(format!("Disk ID mismatch wanted {want_id:?}, got {stored_disk_id:?}")));
+        }
+
+        Ok(())
+    }
+
     /// Check if disk ID is stale
     async fn check_disk_stale(&self) -> Result<()> {
         let Some(current_disk_id) = *self.disk_id.read().await else {

crates/ecstore/src/disk/error_conv.rs

@@ -48,7 +48,6 @@ pub fn to_volume_error(io_err: std::io::Error) -> std::io::Error {
     }
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn to_disk_error(io_err: std::io::Error) -> std::io::Error {
     match io_err.kind() {
         std::io::ErrorKind::NotFound => DiskError::DiskNotFound.into(),

crates/ecstore/src/disk/fs.rs

@@ -178,7 +178,6 @@ pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
     }
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
     // Try remove_file first; fall back to remove_dir_all if it's a directory
     match fs::remove_file(path.as_ref()).await {

crates/ecstore/src/disk/local.rs

@@ -665,7 +665,6 @@ async fn remove_empty_directory_tree_under_mount_lease(
 }
 
 #[cfg(unix)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 async fn remove_empty_directory_tree_with(
     root: &Path,
     before_descend: impl FnMut(&Path) -> std::io::Result<()>,
@@ -1017,29 +1016,13 @@ fn record_direct_read_page_fault_delta(path: &'static str, stage: &'static str,
 /// When enabled, shard reads bypass the page cache using O_DIRECT flag.
 /// Requires aligned buffers (typically 512 bytes or 4096 bytes).
 /// Default: false (uses page cache via mmap/pread).
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE";
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: bool = false;
 
 /// Minimum shard size threshold for O_DIRECT reads.
 /// Only shards larger than this threshold will use O_DIRECT.
 /// Default: 4MB.
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD";
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
 
 /// Enable O_DIRECT for erasure shard / multipart part data writes (Linux only).
@@ -1053,15 +1036,7 @@ const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
 /// EINVAL/EOPNOTSUPP (tmpfs, overlayfs, 9p, ...) latch the path off and fall
 /// back to buffered writes for the whole disk. Non-Linux always falls back.
 /// Default: false (buffered writes via the page cache, as before).
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE";
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: bool = false;
 const ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: &str = "RUSTFS_OBJECT_MMAP_POPULATE_ENABLE";
 const DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: bool = false;
@@ -1120,14 +1095,12 @@ macro_rules! cached_read_env {
 
 cached_read_env! {
     /// Check if O_DIRECT reads are enabled.
-    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn is_direct_io_read_enabled() -> bool =
         rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE);
 }
 
 cached_read_env! {
     /// Check if O_DIRECT shard/part data writes are enabled.
-    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn is_direct_io_write_enabled() -> bool =
         rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE);
 }
@@ -1483,7 +1456,6 @@ pub(crate) fn effective_durability(volume: &str) -> DurabilityMode {
 
 cached_read_env! {
     /// Get the O_DIRECT read threshold size.
-    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn get_direct_io_read_threshold() -> usize =
         rustfs_utils::get_env_usize(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD);
 }
@@ -1701,20 +1673,12 @@ impl DirectIoWriteState {
 /// Target staging size for O_DIRECT writes, rounded up to the DIO alignment.
 /// Bounds the per-writer aligned bounce buffer and batches many shard blocks
 /// into one positioned write to keep the syscall count low.
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const DIRECT_WRITE_STAGING_BYTES: usize = 1024 * 1024;
 
 /// Aligned bounce-buffer capacity for a given DIO alignment: the target staging
 /// size rounded up to a whole multiple of `align` so the buffer address, every
 /// flushed batch length, and every write offset stay alignment-correct.
 /// Platform-independent (no O_DIRECT), so it is unit-tested on any host.
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 fn direct_write_staging_capacity(align: usize) -> usize {
     debug_assert!(align.is_power_of_two() && align >= 512);
     DIRECT_WRITE_STAGING_BYTES.div_ceil(align) * align
@@ -1723,10 +1687,6 @@ fn direct_write_staging_capacity(align: usize) -> usize {
 /// Split `filled` staged bytes into the alignment-sized prefix written with
 /// O_DIRECT and the sub-alignment tail written buffered. Platform-independent,
 /// so the tail-boundary math is unit-tested on any host.
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 fn direct_write_tail_split(filled: usize, align: usize) -> (usize, usize) {
     let aligned = filled - (filled % align);
     (aligned, filled - aligned)
@@ -2182,7 +2142,6 @@ fn set_delete_version_fail_after_data_staged(path: &str) {
 }
 
 #[cfg(test)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(crate) fn set_delete_version_fail_after_commit(root: &Path, path: &str) {
     DELETE_VERSION_FAIL_AFTER_COMMIT
         .lock()
@@ -2488,10 +2447,6 @@ enum SyncMode {
     FileOnly,
 }
 
-#[allow(
-    dead_code,
-    reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
-)]
 struct FileCacheReclaimWriter {
     inner: File,
     reclaim_len: usize,
@@ -2499,10 +2454,6 @@ struct FileCacheReclaimWriter {
     reclaimed: bool,
 }
 
-#[allow(
-    dead_code,
-    reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
-)]
 struct FileCacheReclaimReader {
     inner: File,
     reclaim_offset: u64,
@@ -2568,10 +2519,6 @@ impl<R: AsyncRead + Unpin> AsyncRead for StallTimeoutReader<R> {
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "reclaim metrics emitter reached only from the Linux-gated reclaim paths (backlog#1823)"
-)]
 fn record_file_cache_reclaim_success(kind: &'static str, reclaim_len: usize, started: std::time::Instant) {
     // Runs per read-stream page-cache reclaim window; skip the whole emission
     // (three metric-key constructions) when general metrics are disabled.
@@ -3124,7 +3071,6 @@ impl LocalIoBackend for StdBackend {
             use memmap2::MmapOptions;
             use std::time::{Duration as StdDuration, Instant as StdInstant};
 
-            #[allow(dead_code, reason = "mmap copy result slot kept beside the mapping it owns (backlog#1823)")]
             struct MmapCopyReadResult {
                 bytes: Bytes,
                 access_check_duration: StdDuration,
@@ -4758,10 +4704,6 @@ fn build_local_io_backend(root: PathBuf) -> Arc<dyn LocalIoBackend> {
     Arc::new(StdBackend::new(root))
 }
 
-#[allow(
-    dead_code,
-    reason = "path cache and cwd slots retained beside the disk root they derive from (backlog#1823)"
-)]
 pub struct LocalDisk {
     pub root: PathBuf,
     publication_root: os::PublicationRoot,
@@ -5548,7 +5490,6 @@ impl LocalDisk {
         Ok(Self::resolve_abs_path_from(&self.root, path.as_ref()))
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn io_resolve_abs_path(&self, path: impl AsRef<Path>) -> PathBuf {
         let path_ref = path.as_ref();
         let path_str = path_ref.to_string_lossy();
@@ -5626,24 +5567,15 @@ impl LocalDisk {
     }
 
     // Check if a path is valid
-    #[allow(
-        dead_code,
-        reason = "method wrapper over the live free function check_local_disk_valid_path; no caller in this port (backlog#1823)"
-    )]
     fn check_valid_path<P: AsRef<Path>>(&self, path: P) -> Result<()> {
         check_local_disk_valid_path(self.io_root(), path)
     }
 
-    #[allow(
-        dead_code,
-        reason = "method wrapper over the live free function reject_local_disk_symlink_components; no caller in this port (backlog#1823)"
-    )]
     fn reject_symlink_components(&self, path: &Path) -> Result<()> {
         reject_local_disk_symlink_components(self.io_root(), path)
     }
 
     // Batch path generation with single lock acquisition
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result<Vec<PathBuf>> {
         let mut results = Vec::with_capacity(requests.len());
         let mut cache_misses = Vec::new();
@@ -6556,7 +6488,6 @@ impl LocalDisk {
         Ok(f)
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn open_file_read_only(&self, path: impl AsRef<Path>) -> Result<File> {
         let f = super::fs::open_file(path.as_ref(), O_RDONLY).await.map_err(to_file_error)?;
         Ok(f)

crates/ecstore/src/disk/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 // #730: disk abstractions still carry staged health and direct-I/O migration paths.
+#![allow(dead_code)]
 
 pub mod disk_store;
 pub mod endpoint;
@@ -1113,10 +1114,6 @@ pub struct DiskInfo {
 }
 
 #[derive(Clone, Debug, Default)]
-#[allow(
-    dead_code,
-    reason = "MinIO-parity disk info shape with no constructor in this port (backlog#1823)"
-)]
 pub struct Info {
     pub total: u64,
     pub free: u64,
@@ -1375,7 +1372,6 @@ pub fn conv_part_err_to_int(err: &Option<Error>) -> usize {
     }
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn has_part_err(part_errs: &[usize]) -> bool {
     part_errs.iter().any(|err| *err != CHECK_PART_SUCCESS)
 }

crates/ecstore/src/disk/os.rs

@@ -571,10 +571,6 @@ fn regular_files(dir: &Path) -> io::Result<Vec<PathBuf>> {
 
 /// Fdatasync every regular file directly inside `dir`, then fsync the directory
 /// itself.
-#[allow(
-    dead_code,
-    reason = "reached only through sync_dir_files, whose callers are tests (backlog#1823)"
-)]
 pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
     for entry in std::fs::read_dir(dir.as_ref())? {
         let entry = entry?;
@@ -587,7 +583,6 @@ pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
 
 /// Async wrapper around [`sync_dir_files_std`]. Large directories flush files
 /// concurrently, bounded both per directory and process-wide.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub async fn sync_dir_files(dir: impl AsRef<Path>) -> io::Result<()> {
     sync_dir_files_with_limiter(dir, Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS))).await
 }
@@ -1814,6 +1809,10 @@ impl RenameCommitGuard {
         })
     }
 
+    pub(crate) fn lock_destination_directory_for_path_access(&self, directory: &Path) -> io::Result<RenameDestinationPathGuard> {
+        self.destination_directory_guard(directory, false)
+    }
+
     pub(crate) fn create_destination_directory_for_path_access(
         &self,
         directory: &Path,
@@ -2859,6 +2858,13 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
     Ok(())
 }
 
+/// Check if a file exists.
+/// Returns true if the file exists, false otherwise.
+#[tracing::instrument(level = "debug", skip_all)]
+pub fn file_exists(path: impl AsRef<Path>) -> bool {
+    std::fs::metadata(path.as_ref()).map(|_| true).unwrap_or(false)
+}
+
 /// Whether an [`io::Error`] means "the directory is not empty".
 ///
 /// POSIX lets `rmdir`/`rename` report a non-empty directory as either

crates/ecstore/src/io_support/bitrot.rs

@@ -704,7 +704,6 @@ pub(crate) async fn create_bitrot_reader_from_bytes_with_stage_metrics(
 }
 
 #[allow(clippy::too_many_arguments)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn create_deferred_bitrot_reader(
     inline_data: Option<Bytes>,
     disk: Option<DiskStore>,

crates/ecstore/src/object_api/types.rs

@@ -277,12 +277,6 @@ pub struct ObjectOptions {
     /// fence avoids recursively acquiring the read lock behind a queued writer.
     pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
     pub replication_request: bool,
-    /// Source-cluster LWW timestamps carried by an authorized replication
-    /// request; None when the source never modified the category. Only the
-    /// replication-authorized options builders may set these.
-    pub replication_tagging_timestamp: Option<OffsetDateTime>,
-    pub replication_retention_timestamp: Option<OffsetDateTime>,
-    pub replication_legalhold_timestamp: Option<OffsetDateTime>,
     /// Authorized SSE-C replication passthrough: the body is already
     /// ciphertext, so the write path must not encrypt or compress it and
     /// stores the restored encryption metadata verbatim. Only the

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -32,22 +32,15 @@ use crate::diagnostics::get::{
     GET_METADATA_CACHE_REASON_NOT_READ_DATA, GET_METADATA_CACHE_REASON_PART_NUMBER,
     GET_METADATA_CACHE_REASON_RAW_DATA_MOVEMENT_READ, GET_METADATA_CACHE_REASON_USABLE, GET_METADATA_CACHE_REASON_VERSION_ID,
     GET_METADATA_CACHE_REASON_VERSION_SUSPENDED, GET_METADATA_CACHE_REASON_VERSIONED,
-    GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY,
+    GET_METADATA_EARLY_STOP_REASON_CONFLICTING_METADATA, GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY,
+    GET_METADATA_EARLY_STOP_REASON_ERROR, GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
+    GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
+    GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE,
+    GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND, GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE,
+    GET_METADATA_RESPONSE_ERROR, GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED,
+    GET_METADATA_RESPONSE_VALID, GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING,
-    GET_METADATA_EARLY_STOP_REASON_DELETE_MARKER, GET_METADATA_EARLY_STOP_REASON_ERROR,
+    GET_OBJECT_PATH_DIRECT_MEMORY, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK,
-    GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM, GET_METADATA_EARLY_STOP_REASON_NOT_FOUND,
+    GET_STAGE_DECODE, GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
-    GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM,
-    GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM, GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND,
-    GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND, GET_METADATA_RESPONSE_ERROR,
-    GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT, GET_METADATA_RESPONSE_VALID,
-    GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_DIRECT_MEMORY,
-    GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE,
-    GET_STAGE_METADATA_CACHE_LOOKUP, GET_STAGE_METADATA_RESOLVE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP,
     GET_STAGE_READER_SETUP_DROP_PENDING, GET_STAGE_READER_SETUP_SCHEDULE, GET_STAGE_READER_SETUP_WAIT_QUORUM,
     GET_STAGE_READER_TASK_BITROT_READER_INIT, GET_STAGE_READER_TASK_FILE_OPEN, GET_STAGE_READER_TASK_READER_CONSTRUCTION,
     GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
@@ -180,13 +173,11 @@ pub(in crate::set_disk) enum GetCodecStreamingReaderBuildOutcome {
     Fallback(GetCodecStreamingFallbackReason),
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) struct MultipartCodecStreamingReader {
     pub(in crate::set_disk) readers: VecDeque<Box<dyn AsyncRead + Unpin + Send + Sync>>,
 }
 
 impl MultipartCodecStreamingReader {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) fn new(readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>>) -> Self {
         Self {
             readers: VecDeque::from(readers),
@@ -661,15 +652,36 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
         && left.erasure.distribution == right.erasure.distribution
 }
 
-pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
+pub(in crate::set_disk) async fn data_read_early_stop_inline_body_verified(
     bucket: &str,
     object: &str,
     candidate: &FileInfo,
     parts_metadata: &[FileInfo],
     disks: &[Option<DiskStore>],
-) -> Option<&'static str> {
+) -> bool {
-    if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) {
+    if !candidate.inline_data()
-        return Some(reason);
+        || candidate.is_compressed()
+        || candidate
+            .metadata
+            .keys()
+            .any(|key| rustfs_utils::http::is_object_encryption_marker(key))
+        || candidate.is_remote()
+        || candidate.deleted
+        || candidate.size <= 0
+        || candidate.parts.len() != 1
+        || !candidate.has_valid_erasure_geometry()
+    {
+        return false;
+    }
+
+    let Ok(object_size) = usize::try_from(candidate.size) else {
+        return false;
+    };
+    if candidate.parts.first().is_none_or(|part| part.size != object_size) {
+        return false;
+    }
+    if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
+        return false;
     }
 
     let Ok(erasure) = coding::Erasure::try_new_with_options(
@@ -678,21 +690,18 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
         candidate.erasure.block_size,
         candidate.uses_legacy_checksum,
     ) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+        return false;
     };
-    let data_files =
+    let Some(data_files) =
-        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
+        collect_inline_data_shard_fileinfos_by_index(parts_metadata, candidate, erasure.data_shards, |index| {
             disks.get(index).is_some_and(Option::is_some)
-        }) {
+        })
-            Ok(data_files) => data_files,
+    else {
-            Err(reason) => return Some(reason),
+        return false;
-        };
+    };
 
     let Some(part) = candidate.parts.first() else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
+        return false;
-    };
-    let Ok(object_size) = usize::try_from(candidate.size) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
     };
     let checksum_info = candidate.erasure.get_checksum_info(part.number);
     let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
@@ -712,111 +721,12 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
     let Ok(mut readers) =
         build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
     else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
-    };
-
-    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
-        Some(body) if body.len() == object_size => None,
-        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
-    }
-}
-
-fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Option<&'static str> {
-    // `inline_data` excludes remote objects; this diagnostic reports them separately.
-    if !rustfs_utils::http::contains_key_str(&candidate.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA) {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
-    }
-    if candidate.is_compressed()
-        || candidate
-            .metadata
-            .keys()
-            .any(|key| rustfs_utils::http::is_object_encryption_marker(key))
-    {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED);
-    }
-    if candidate.is_remote() {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE);
-    }
-    if candidate.deleted {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED);
-    }
-    if candidate.size <= 0 {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
-    }
-    if candidate.parts.len() != 1 {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
-    }
-    if !candidate.has_valid_erasure_geometry() {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
-    }
-
-    let Ok(object_size) = usize::try_from(candidate.size) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
-    };
-    if candidate.parts.first().is_none_or(|part| part.size != object_size) {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
-    }
-    if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
-    }
-    None
-}
-
-fn data_read_inline_missing_shards_are_pending(
-    candidate: &FileInfo,
-    parts_metadata: &[FileInfo],
-    errors: &[Option<DiskError>],
-    disks: &[Option<DiskStore>],
-    fanout_order: &[usize],
-    scheduled_fanout_len: usize,
-) -> bool {
-    let Ok(erasure) = coding::Erasure::try_new_with_options(
-        candidate.erasure.data_blocks,
-        candidate.erasure.parity_blocks,
-        candidate.erasure.block_size,
-        candidate.uses_legacy_checksum,
-    ) else {
         return false;
     };
-    let distribution = &candidate.erasure.distribution;
-    let mut data_shards_seen_or_pending = vec![false; erasure.data_shards];
-    let mut missing_pending_data_shards = 0usize;
 
-    for (disk_index, file_info) in parts_metadata.iter().enumerate() {
+    try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size)
-        let Some(&block_index) = distribution.get(disk_index) else {
+        .await
-            return false;
+        .is_some_and(|body| body.len() == object_size)
-        };
-        if block_index == 0 || block_index > erasure.data_shards {
-            continue;
-        }
-        if !disks.get(disk_index).is_some_and(Option::is_some) {
-            return false;
-        }
-
-        let data_slot = block_index - 1;
-        if file_info.name.is_empty() {
-            let scheduled_and_not_failed = fanout_order
-                .get(..scheduled_fanout_len)
-                .is_some_and(|scheduled_disks| scheduled_disks.contains(&disk_index))
-                && errors.get(disk_index).is_some_and(Option::is_none);
-            if scheduled_and_not_failed {
-                data_shards_seen_or_pending[data_slot] = true;
-                missing_pending_data_shards = missing_pending_data_shards.saturating_add(1);
-                continue;
-            }
-            return false;
-        }
-        if file_info.erasure.index != block_index
-            || !file_info.has_valid_erasure_geometry()
-            || !metadata_early_stop_candidate_matches(file_info, candidate)
-            || file_info.data.as_ref().is_none_or(|data| data.is_empty())
-        {
-            return false;
-        }
-        data_shards_seen_or_pending[data_slot] = true;
-    }
-
-    missing_pending_data_shards > 0 && data_shards_seen_or_pending.into_iter().all(|seen_or_pending| seen_or_pending)
 }
 
 pub(in crate::set_disk) fn classify_metadata_response_error(err: &DiskError) -> &'static str {
@@ -1848,7 +1758,6 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_all_shards(
 }
 
 #[allow(clippy::too_many_arguments)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum(
     files: &[FileInfo],
     disks: &[Option<DiskStore>],
@@ -2139,7 +2048,6 @@ pub(in crate::set_disk) async fn create_data_block_bitrot_readers(
     setup
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) async fn collect_read_multiple_results<F>(
     tasks: Vec<F>,
     read_quorum: usize,
@@ -2456,7 +2364,6 @@ impl SetDisks {
         let bucket: Arc<str> = Arc::from(bucket);
         let object: Arc<str> = Arc::from(object);
         let version_id: Arc<str> = Arc::from(version_id);
-        let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
         let futures = disks.iter().enumerate().map(|(disk_index, disk)| {
             let disk = disk.clone();
             let task_opts = opts;
@@ -2464,14 +2371,10 @@ impl SetDisks {
             let bucket = bucket.clone();
             let object = object.clone();
             let version_id = version_id.clone();
-            let slowtail_fault = slowtail_fault.clone();
             tokio::spawn(async move {
                 let response_start = observe.then(Instant::now);
                 let result = if let Some(disk) = disk {
                     Self::record_read_version_call(&object, disk_index);
-                    if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(disk_index)) {
-                        tokio::time::sleep(delay).await;
-                    }
                     disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
                         .await
                 } else {
@@ -2566,8 +2469,6 @@ impl SetDisks {
         let mut next_fanout_index = 0usize;
         let mut scheduled_count = 0usize;
         let mut force_full_wait = false;
-        let mut final_miss_reason_override = None;
-        let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
         let spawn_read_version =
             |join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
                 let task_opts = opts;
@@ -2575,7 +2476,6 @@ impl SetDisks {
                 let bucket = bucket.clone();
                 let object = object.clone();
                 let version_id = version_id.clone();
-                let slowtail_fault = slowtail_fault.clone();
                 join_set.spawn(async move {
                     let response_start = Instant::now();
                     let result = if let Some(disk) = disk {
@@ -2584,9 +2484,6 @@ impl SetDisks {
                         Self::record_read_version_call(&object, index);
                         #[cfg(test)]
                         Self::read_version_fanout_barrier(&object, index).await;
-                        if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(index)) {
-                            tokio::time::sleep(delay).await;
-                        }
                         disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
                             .await
                     } else {
@@ -2614,20 +2511,11 @@ impl SetDisks {
         }
 
         while let Some(result) = join_set.join_next().await {
-            let mut defer_pending_inline_data_shard = false;
             match result {
                 Ok((index, res, elapsed)) => match res {
                     Ok(file_info) => {
                         observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
                         accumulator.observe_file_info(&file_info);
-                        if bounded_fanout
-                            && read_data
-                            && !force_full_wait
-                            && let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
-                        {
-                            force_full_wait = true;
-                            final_miss_reason_override.get_or_insert(reason);
-                        }
                         if let Some(slot) = ress.get_mut(index) {
                             *slot = file_info;
                         }
@@ -2653,43 +2541,17 @@ impl SetDisks {
                     .or_else(|| accumulator.version_early_stop_decision())
             {
                 let should_return_early = if read_data {
-                    match accumulator.candidate.as_ref() {
+                    let allow_data_read_early_stop = match accumulator.candidate.as_ref() {
-                        Some(candidate) => match data_read_early_stop_inline_body_miss_reason(
+                        Some(candidate) => {
-                            bucket.as_ref(),
+                            data_read_early_stop_inline_body_verified(bucket.as_ref(), object.as_ref(), candidate, &ress, disks)
-                            object.as_ref(),
+                                .await
-                            candidate,
-                            &ress,
-                            disks,
-                        )
-                        .await
-                        {
-                            None => true,
-                            Some(reason) => {
-                                final_miss_reason_override = Some(reason);
-                                if bounded_fanout
-                                    && reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD
-                                    && data_read_inline_missing_shards_are_pending(
-                                        candidate,
-                                        &ress,
-                                        &errors,
-                                        disks,
-                                        &fanout_order,
-                                        next_fanout_index,
-                                    )
-                                {
-                                    defer_pending_inline_data_shard = true;
-                                } else {
-                                    force_full_wait = true;
-                                }
-                                false
-                            }
-                        },
-                        None => {
-                            force_full_wait = true;
-                            final_miss_reason_override = Some(GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM);
-                            false
                         }
+                        None => false,
+                    };
+                    if !allow_data_read_early_stop {
+                        force_full_wait = true;
                     }
+                    allow_data_read_early_stop
                 } else {
                     true
                 };
@@ -2726,7 +2588,6 @@ impl SetDisks {
             let pending_responses = join_set.len();
             let should_hedge_single_pending_data_read = read_data
                 && !force_full_wait
-                && !defer_pending_inline_data_shard
                 && pending_responses == 1
                 && accumulator.can_still_reach_early_stop_with_pending(pending_responses);
             if bounded_fanout && force_full_wait {
@@ -2739,7 +2600,6 @@ impl SetDisks {
                     next_fanout_index = next_fanout_index.saturating_add(1);
                 }
             } else if bounded_fanout
-                && !defer_pending_inline_data_shard
                 && next_fanout_index < disks.len()
                 && (!accumulator.can_still_reach_early_stop_with_pending(pending_responses)
                     || should_hedge_single_pending_data_read)
@@ -2753,12 +2613,7 @@ impl SetDisks {
             }
         }
 
-        let accumulator_miss_reason = accumulator.final_miss_reason();
+        rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, accumulator.final_miss_reason());
-        let final_miss_reason = match (final_miss_reason_override, accumulator_miss_reason) {
-            (Some(reason), GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM) => reason,
-            _ => accumulator_miss_reason,
-        };
-        rustfs_io_metrics::record_get_object_metadata_early_stop_miss(metrics_path, final_miss_reason);
         rustfs_io_metrics::record_get_object_metadata_early_stop_saved_responses(metrics_path, 0);
         rustfs_io_metrics::record_get_object_metadata_fanout_lifecycle(metrics_path, scheduled_count, scheduled_count, 0);
         let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations);
@@ -2987,7 +2842,6 @@ impl SetDisks {
         (meta_file_infos, errs)
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn read_multiple_files(
         disks: &[Option<DiskStore>],
         req: ReadMultipleReq,
@@ -3021,11 +2875,14 @@ impl SetDisks {
             });
         }
 
-        let (ress, _errors) = match collect_read_multiple_results(futures, read_quorum).await {
+        let (ress, errors) = match collect_read_multiple_results(futures, read_quorum).await {
             Ok(collected) => collected,
             Err(()) => return empty_quorum_result(),
         };
 
+        // debug!("ReadMultipleResp ress {:?}", ress);
+        // debug!("ReadMultipleResp errors {:?}", errors);
+
         let mut ret = Vec::with_capacity(req.files.len());
 
         for want in req.files.iter() {
@@ -3164,7 +3021,6 @@ pub(in crate::set_disk) struct RenameDataCommit {
     pub(in crate::set_disk) committed_file_info: FileInfo,
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 type RenameDataLegacyTuple = (
     Vec<Option<DiskStore>>,
     RenameConvergence,
@@ -3174,7 +3030,6 @@ type RenameDataLegacyTuple = (
 );
 
 impl RenameDataCommit {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn into_legacy_tuple(self) -> RenameDataLegacyTuple {
         (
             self.online_disks,
@@ -3293,7 +3148,6 @@ impl SetDisks {
 
     #[tracing::instrument(level = "debug", skip(disks, file_infos))]
     #[allow(clippy::type_complexity)]
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn rename_data(
         disks: &[Option<DiskStore>],
         src_bucket: &str,
@@ -5106,7 +4960,6 @@ fn is_cleanup_not_found(e: &DiskError) -> bool {
 /// normalized to `DiskNotFound`: a panic is not a "disk absent" condition and
 /// must not be silently swallowed as an ignorable error (fixes the historical
 /// `Unexpected`/`DiskNotFound` misclassification).
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn map_cleanup_join_result(joined: std::result::Result<Option<DiskError>, tokio::task::JoinError>) -> Option<DiskError> {
     match joined {
         Ok(res) => res,
@@ -5331,7 +5184,6 @@ pub(in crate::set_disk) mod rename_fanout_barrier_phase {
     /// The per-disk old-data-dir cleanup phase of the commit fan-out.
     pub const CLEANUP: &str = "cleanup";
     /// The per-disk `read_version` phase of metadata read fan-out.
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub const READ_VERSION: &str = "read_version";
 }
 
@@ -5769,130 +5621,6 @@ mod tests {
         (dirs, disks)
     }
 
-    #[test]
-    fn metadata_slowtail_fault_delay_parses_and_filters_request() {
-        temp_env::with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,3")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some("bench-bucket")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
-            ],
-            || {
-                assert_eq!(
-                    get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, true),
-                    Some(Duration::from_millis(25))
-                );
-                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 2, true).is_none());
-                assert!(get_metadata_slowtail_fault_delay("other-bucket", "objects/000001", 3, true).is_none());
-                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "other/000001", 3, true).is_none());
-                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, false).is_none());
-            },
-        );
-    }
-
-    #[test]
-    fn metadata_slowtail_fault_delay_disables_invalid_disk_list() {
-        temp_env::with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,nope")),
-            ],
-            || {
-                assert!(get_metadata_slowtail_fault_delay("bucket", "object", 1, true).is_none());
-            },
-        );
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn metadata_slowtail_fault_delays_only_data_read_metadata_task() {
-        const DISKS: usize = 4;
-        let bucket = "metadata-slowtail-fault-bucket";
-        let object = "objects/metadata-slowtail-fault-object";
-        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
-        install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
-
-        temp_env::async_with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("false")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
-            ],
-            async {
-                let read_without_data =
-                    SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", false, false, false, true, 2);
-                tokio::time::timeout(Duration::from_millis(100), read_without_data)
-                    .await
-                    .expect("non-data metadata fanout must not be delayed by the data-read slowtail hook")
-                    .expect("metadata fanout without read_data should resolve");
-
-                let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
-                    &disks, bucket, bucket, object, "", true, false, false, true, 2,
-                ));
-                assert!(
-                    tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
-                        .await
-                        .is_err(),
-                    "data-read metadata fanout must wait for the injected slow read_version response"
-                );
-                let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
-                    .await
-                    .expect("injected slowtail should eventually complete")
-                    .expect("data-read metadata fanout should resolve");
-                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
-                assert!(errs.iter().all(Option::is_none));
-                assert_eq!(diagnostics.total_responses(), DISKS);
-            },
-        )
-        .await;
-
-        drop(dirs);
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn metadata_slowtail_fault_delays_early_stop_metadata_task() {
-        const DISKS: usize = 4;
-        let bucket = "metadata-slowtail-early-stop-bucket";
-        let object = "objects/metadata-slowtail-early-stop-object";
-        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
-        install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
-
-        temp_env::async_with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
-                (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")),
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
-            ],
-            async {
-                let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
-                    &disks, bucket, bucket, object, "", true, false, false, true, 2,
-                ));
-                assert!(
-                    tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
-                        .await
-                        .is_err(),
-                    "early-stop metadata fanout must still wait for the injected slow response after fallback to full wait"
-                );
-                let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
-                    .await
-                    .expect("injected early-stop slowtail should eventually complete")
-                    .expect("early-stop metadata fanout should resolve");
-                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
-                assert!(errs.iter().all(Option::is_none));
-                assert_eq!(diagnostics.total_responses(), DISKS);
-            },
-        )
-        .await;
-
-        drop(dirs);
-    }
-
     /// Demo / regression guard for the backlog#1325 per-disk call counters.
     ///
     /// The metadata fan-out issues each `read_version` inside its own
@@ -6024,20 +5752,9 @@ mod tests {
         object: &str,
         payload: &[u8],
         uses_legacy_checksum: bool,
-    ) -> Vec<FileInfo> {
-        inline_metadata_fanout_fileinfos_with_geometry(bucket, object, payload, uses_legacy_checksum, 2, 2).await
-    }
-
-    async fn inline_metadata_fanout_fileinfos_with_geometry(
-        bucket: &str,
-        object: &str,
-        payload: &[u8],
-        uses_legacy_checksum: bool,
-        data_shards: usize,
-        parity_shards: usize,
     ) -> Vec<FileInfo> {
         let distribution_key = metadata_distribution_key(bucket, object);
-        let mut base = FileInfo::new(&distribution_key, data_shards, parity_shards);
+        let mut base = FileInfo::new(&distribution_key, 2, 2);
         base.volume = bucket.to_string();
         base.name = object.to_string();
         base.size = i64::try_from(payload.len()).expect("test payload should fit i64");
@@ -6100,21 +5817,6 @@ mod tests {
         install_inline_metadata_fanout_files(disks, bucket, object, files).await;
     }
 
-    async fn install_inline_metadata_fanout_fileinfo_with_geometry(
-        disks: &[Option<DiskStore>],
-        bucket: &str,
-        object: &str,
-        payload: &[u8],
-        data_shards: usize,
-        parity_shards: usize,
-        mutate: impl FnOnce(&mut [FileInfo]),
-    ) {
-        let mut files =
-            inline_metadata_fanout_fileinfos_with_geometry(bucket, object, payload, false, data_shards, parity_shards).await;
-        mutate(&mut files);
-        install_inline_metadata_fanout_files(disks, bucket, object, files).await;
-    }
-
     async fn install_inline_metadata_fanout_files(disks: &[Option<DiskStore>], bucket: &str, object: &str, files: Vec<FileInfo>) {
         let distribution = files
             .first()
@@ -6335,118 +6037,6 @@ mod tests {
         drop(dirs);
     }
 
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn bounded_metadata_early_stop_waits_for_pending_inline_data_shard() {
-        const DISKS: usize = 6;
-        const DATA_SHARDS: usize = 4;
-        const PARITY_SHARDS: usize = 2;
-        let bucket = "bounded-inline-data-get-pending-shard-bucket";
-        let object =
-            object_with_initial_data_shards(bucket, "bounded-inline-data-get-pending-shard-object", DATA_SHARDS, DATA_SHARDS);
-        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
-        install_inline_metadata_fanout_fileinfo_with_geometry(
-            &disks,
-            bucket,
-            &object,
-            b"verified inline payload",
-            DATA_SHARDS,
-            PARITY_SHARDS,
-            |_| {},
-        )
-        .await;
-
-        temp_env::async_with_vars(
-            [
-                ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
-                ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
-                ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
-            ],
-            async {
-                let fanout_order = bounded_metadata_fanout_order(bucket, &object, DISKS, PARITY_SHARDS);
-                let distribution_key = metadata_distribution_key(bucket, &object);
-                let distribution = FileInfo::new(&distribution_key, DATA_SHARDS, PARITY_SHARDS)
-                    .erasure
-                    .distribution;
-                let paused_data_disk = *fanout_order
-                    .iter()
-                    .take(DATA_SHARDS)
-                    .find(|disk_index| {
-                        distribution
-                            .get(**disk_index)
-                            .is_some_and(|block_index| (1..=DATA_SHARDS).contains(block_index))
-                    })
-                    .expect("initial fanout should include a data shard to pause");
-                let hedged_parity_disk = fanout_order[DATA_SHARDS];
-                let unscheduled_parity_disk = fanout_order[DATA_SHARDS + 1];
-
-                let barrier = rename_fanout_barrier::arm(&object, paused_data_disk, rename_fanout_barrier::PHASE_READ_VERSION);
-                let tracker = rename_fanout_barrier::observe_tasks(&object);
-                let calls = disk_call_counters::observe(&object);
-                let disks_for_read = disks.clone();
-                let object_for_read = object.clone();
-                let mut read = tokio::spawn(async move {
-                    SetDisks::read_all_fileinfo_observed(
-                        &disks_for_read,
-                        bucket,
-                        bucket,
-                        &object_for_read,
-                        "",
-                        true,
-                        false,
-                        false,
-                        true,
-                        PARITY_SHARDS,
-                    )
-                    .await
-                });
-
-                tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
-                    .await
-                    .expect("initial data shard should pause before returning");
-                tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
-                    while calls.for_disk(disk_call_counters::KIND_READ_VERSION, hedged_parity_disk) == 0 {
-                        tokio::task::yield_now().await;
-                    }
-                })
-                .await
-                .expect("bounded fanout should hedge one parity disk while the data shard is pending");
-
-                assert!(
-                    tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await.is_err(),
-                    "inline data-read early-stop must wait for a scheduled missing data shard instead of forcing full wait"
-                );
-
-                barrier.release();
-                let (parts_metadata, errs, diagnostics) = read
-                    .await
-                    .expect("metadata read task should not panic")
-                    .expect("pending data shard should let the inline verifier finish");
-
-                assert_eq!(
-                    calls.total(disk_call_counters::KIND_READ_VERSION),
-                    5,
-                    "pending data-shard defer should not schedule the final parity disk"
-                );
-                assert_eq!(
-                    calls.for_disk(disk_call_counters::KIND_READ_VERSION, unscheduled_parity_disk),
-                    0,
-                    "the remaining parity disk must stay unissued when pending data verification succeeds"
-                );
-                assert_eq!(
-                    tracker.running(),
-                    0,
-                    "early-stop should drain spawned read_version tasks before returning"
-                );
-                assert_eq!(diagnostics.total_responses(), 5);
-                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), 5);
-                assert!(errs.iter().all(Option::is_none));
-            },
-        )
-        .await;
-
-        drop(dirs);
-    }
-
     #[tokio::test]
     async fn data_read_early_stop_verifies_legacy_inline_checksum_payload() {
         let bucket = "legacy-inline-data-get-fanout-bucket";
@@ -6477,133 +6067,11 @@ mod tests {
             .clone();
 
         assert!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &parts_metadata, &disks)
+            data_read_early_stop_inline_body_verified(bucket, object, &candidate, &parts_metadata, &disks).await,
-                .await
-                .is_none(),
             "legacy inline metadata must use the legacy bitrot shard sizing and checksum algorithm"
         );
     }
 
-    #[tokio::test]
-    async fn data_read_early_stop_reports_inline_miss_reasons() {
-        let bucket = "inline-data-get-miss-reason-bucket";
-        let object = "inline-data-get-miss-reason-object";
-        let payload = b"verified inline payload";
-        let (_dirs, disks) = call_counter_local_disks(bucket, 4).await;
-        let files = inline_metadata_fanout_fileinfos_with_mode(bucket, object, payload, false).await;
-        let distribution = files
-            .first()
-            .map(|file| file.erasure.distribution.clone())
-            .expect("fixture should include metadata");
-        let order = bounded_metadata_fanout_order(bucket, object, 4, 2);
-        let mut parts_metadata = vec![FileInfo::default(); 4];
-        for disk_index in order.into_iter().take(3) {
-            let block_index = distribution
-                .get(disk_index)
-                .copied()
-                .expect("fixture distribution should cover every disk");
-            parts_metadata[disk_index] = files
-                .get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
-                .expect("fixture should include every distributed shard")
-                .clone();
-        }
-        let candidate = parts_metadata
-            .iter()
-            .find(|file| file.name == object)
-            .expect("fixture should include observed metadata")
-            .clone();
-        let data_disk = distribution
-            .iter()
-            .position(|block_index| *block_index == 1)
-            .expect("fixture distribution should include first data shard");
-
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &parts_metadata, &disks).await,
-            None
-        );
-
-        let mut not_inline = candidate.clone();
-        rustfs_utils::http::remove_str(&mut not_inline.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA);
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &not_inline, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE)
-        );
-
-        let mut remote = candidate.clone();
-        remote.transition_status = TRANSITION_COMPLETE.to_string();
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &remote, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_REMOTE)
-        );
-
-        let mut transformed = candidate.clone();
-        rustfs_utils::http::insert_str(&mut transformed.metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &transformed, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_TRANSFORMED)
-        );
-
-        let mut deleted = candidate.clone();
-        deleted.deleted = true;
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &deleted, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_DELETED)
-        );
-
-        let mut zero_size = candidate.clone();
-        zero_size.size = 0;
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &zero_size, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE)
-        );
-
-        let mut multipart = candidate.clone();
-        multipart.parts.push(multipart.parts[0].clone());
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &multipart, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE)
-        );
-
-        let mut invalid_geometry = candidate.clone();
-        invalid_geometry.erasure.data_blocks = 0;
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &invalid_geometry, &parts_metadata, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY)
-        );
-
-        let mut missing_shard = parts_metadata.clone();
-        missing_shard[data_disk] = FileInfo::default();
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &missing_shard, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD)
-        );
-
-        let mut missing_payload = parts_metadata.clone();
-        missing_payload[data_disk].data = None;
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &missing_payload, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD)
-        );
-
-        let mut identity_mismatch = parts_metadata.clone();
-        identity_mismatch[data_disk].version_id = Some(Uuid::new_v4());
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &identity_mismatch, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH)
-        );
-
-        let mut corrupt = parts_metadata.clone();
-        if let Some(data) = corrupt[data_disk].data.as_mut() {
-            let mut corrupt_data = data.to_vec();
-            corrupt_data[0] ^= 0x01;
-            *data = Bytes::from(corrupt_data);
-        }
-        assert_eq!(
-            data_read_early_stop_inline_body_miss_reason(bucket, object, &candidate, &corrupt, &disks).await,
-            Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY)
-        );
-    }
-
     #[test]
     #[serial_test::serial]
     fn metadata_fanout_lifecycle_records_real_early_stop_abort() {
@@ -6693,7 +6161,7 @@ mod tests {
                 &[
                     ("path", GET_OBJECT_PATH_INTERNAL_META),
                     ("decision", "miss"),
-                    ("reason", GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+                    ("reason", GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM),
                 ],
             ),
             1,
@@ -6705,7 +6173,7 @@ mod tests {
                 &[
                     ("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
                     ("decision", "miss"),
-                    ("reason", GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+                    ("reason", GET_METADATA_EARLY_STOP_REASON_INSUFFICIENT_QUORUM),
                 ],
             ),
             0,
@@ -7240,7 +6708,7 @@ mod tests {
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn bounded_non_inline_data_get_immediately_forces_full_fanout() {
+    async fn bounded_non_inline_data_get_hedges_then_waits_for_full_fanout() {
         const DISKS: usize = 4;
         let bucket = "bounded-data-get-hedge-bucket";
         let object = "bounded-data-get-hedge-object";
@@ -7271,7 +6739,7 @@ mod tests {
                     }
                 })
                 .await
-                .expect("bounded non-inline data-read fanout should immediately schedule the spare disk");
+                .expect("bounded data-read fanout should hedge by starting the spare disk");
 
                 let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
                 assert!(
@@ -7287,7 +6755,7 @@ mod tests {
                 assert_eq!(
                     calls.total(disk_call_counters::KIND_READ_VERSION),
                     DISKS as u64,
-                    "bounded non-inline data-read fanout should issue the paused disk plus the remaining spare"
+                    "bounded data-read fanout should issue the paused disk plus one spare hedge"
                 );
                 assert_eq!(diagnostics.total_responses(), DISKS);
                 assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
@@ -7300,7 +6768,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn bounded_metadata_early_stop_defaults_keep_non_inline_data_get_full_fanout() {
+    async fn bounded_metadata_early_stop_defaults_keep_data_get_full_fanout() {
         const DISKS: usize = 4;
         let bucket = "bounded-data-get-default-bucket";
         let object = "bounded-data-get-default-object";
@@ -7314,42 +6782,16 @@ mod tests {
                 ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
             ],
             async {
-                let barrier = rename_fanout_barrier::arm(object, 2, rename_fanout_barrier::PHASE_READ_VERSION);
                 let calls = disk_call_counters::observe(object);
-                let disks_for_read = disks.clone();
+                let (parts_metadata, errs, diagnostics) =
-                let mut read = tokio::spawn(async move {
+                    SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2)
-                    SetDisks::read_all_fileinfo_observed(&disks_for_read, bucket, bucket, object, "", true, false, false, true, 2)
                         .await
-                });
+                        .expect("default data-read metadata should resolve");
-
-                tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
-                    .await
-                    .expect("default bounded non-inline read should schedule the paused metadata task");
-                tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
-                    while calls.for_disk(disk_call_counters::KIND_READ_VERSION, 3) == 0 {
-                        tokio::task::yield_now().await;
-                    }
-                })
-                .await
-                .expect(
-                    "default bounded non-inline read should immediately force full fanout after the first non-inline response",
-                );
-
-                let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
-                assert!(
-                    pending.is_err(),
-                    "default non-inline data reads must not return before the paused metadata response"
-                );
-                barrier.release();
-                let (parts_metadata, errs, diagnostics) = read
-                    .await
-                    .expect("metadata read task should not panic")
-                    .expect("default data-read metadata should resolve");
 
                 assert_eq!(
                     calls.total(disk_call_counters::KIND_READ_VERSION),
                     DISKS as u64,
-                    "default non-inline GET data-read metadata must keep full fanout without waiting for a quorum miss first"
+                    "default GET data-read metadata must keep full fanout for read-failure tolerance"
                 );
                 assert_eq!(diagnostics.total_responses(), DISKS);
                 assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);

crates/ecstore/src/set_disk/ctx.rs

@@ -42,20 +42,12 @@ impl<'a> SetDisksCtx<'a> {
     }
 
     /// The borrowed core, for state not yet fronted by a typed accessor.
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn core(&self) -> &'a SetDisks {
         self.core
     }
 
     // --- Immutable topology / config (fixed after construction) ---
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn set_index(&self) -> usize {
         self.core.set_index
     }
@@ -64,26 +56,14 @@ impl<'a> SetDisksCtx<'a> {
         self.core.pool_index
     }
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn set_drive_count(&self) -> usize {
         self.core.set_drive_count
     }
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn default_parity_count(&self) -> usize {
         self.core.default_parity_count
     }
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn set_endpoints(&self) -> &'a [Endpoint] {
         &self.core.set_endpoints
     }
@@ -92,10 +72,6 @@ impl<'a> SetDisksCtx<'a> {
         &self.core.format
     }
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn locker_owner(&self) -> &'a str {
         &self.core.locker_owner
     }
@@ -108,10 +84,6 @@ impl<'a> SetDisksCtx<'a> {
 
     // --- Locker trio ---
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn lockers(&self) -> &'a [Arc<dyn LockClient>] {
         &self.core.lockers
     }

crates/ecstore/src/set_disk/mod.rs

@@ -39,6 +39,7 @@
 //! - `metadata.rs`, `replication.rs`, `shard_source.rs` — supporting helpers.
 
 // #730: SetDisks still hosts staged read/heal/write migration helpers.
+#![allow(dead_code)]
 #![allow(unused_imports)]
 #![allow(unused_variables)]
 
@@ -58,10 +59,7 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
 use crate::cluster::rpc::heal_bucket_local_on_disks;
 use crate::data_usage::record_compression_total_memory;
 use crate::diagnostics::get::{
-    GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY,
+    GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD,
-    GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING,
     GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_DIRECT_MEMORY,
     GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX,
     GET_OBJECT_PATH_REMOTE_TRANSITION, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE,
@@ -102,7 +100,9 @@ use crate::storage_api_contracts::{
 };
 use crate::store::utils::is_reserved_or_invalid_bucket;
 use crate::{
-    bucket::lifecycle::bucket_lifecycle_ops::{LifecycleOps, get_transitioned_object_reader_with_tier_manager, put_restore_opts},
+    bucket::lifecycle::bucket_lifecycle_ops::{
+        LifecycleOps, gen_transition_objname, get_transitioned_object_reader_with_tier_manager, put_restore_opts,
+    },
     cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
     config::storageclass,
     disk::{
@@ -174,14 +174,15 @@ use std::future::Future;
 use std::hash::{BuildHasher, Hash, Hasher};
 use std::mem::{self};
 use std::pin::Pin;
+use std::sync::OnceLock;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::{Arc, OnceLock};
 use std::task::{Context, Poll};
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 use std::{
     collections::{HashMap, HashSet},
     io::{Cursor, Write},
     path::Path,
+    sync::Arc,
     time::Duration,
 };
 use time::OffsetDateTime;
@@ -620,9 +621,7 @@ fn adaptive_duplex_buffer_size(object_size: i64) -> usize {
 // Each flag has a corresponding `*_ROLLOUT_PCT` for percentage-based gradual rollout.
 // ============================================================================
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
 const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_secs(2); // Increased from 250ms to 2s
 const DEFAULT_GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 4096; // Increased from 1024 to 4096
@@ -690,36 +689,22 @@ const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: usize = 128 * 102
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE";
 // Enabled by default (backlog#872): the early-stop path only engages for
 // requests `should_allow_metadata_early_stop` classifies as safe (latest-version
-// reads by default, without version_id / healing / free-version needs) and still
+// metadata-only reads by default, without version_id / healing / free-version
-// requires a full read-quorum agreement before stopping. Data-read requests add
+// needs) and still requires a full read-quorum agreement before stopping. Set
-// a separate inline-shard verifier before cancelling the remaining fanout. Set
 // the env var to `false` to fall back to full-wait metadata fanout.
 const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true;
 
-#[allow(
-    dead_code,
-    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
-)]
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT";
-#[allow(
-    dead_code,
-    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
-)]
 const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100;
 
 const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE";
 const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false;
 
 const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE";
-const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
+const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = false;
 
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
-const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true;
+const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false;
-
-const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";
-const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS";
-const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET";
-const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX";
 
 // --- Multipart Reader-Setup Prefetch Configuration (backlog#870) ---
 
@@ -921,16 +906,18 @@ mod prepared_get_object_metadata_tests {
             .expect("test should find an object whose initial fanout covers both data shards")
     }
 
-    #[allow(
-        dead_code,
-        reason = "test fixture no assertion in this module uses today; the live namesake lives in io_primitives tests (backlog#1823)"
-    )]
     fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
         *bounded_metadata_fanout_order(bucket, object, 4, 2)
             .get(3)
             .expect("4-disk test geometry should leave one bounded spare disk")
     }
 
+    fn bounded_slow_initial_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(2)
+            .expect("4-disk test geometry should include a third initial metadata disk")
+    }
+
     #[tokio::test]
     async fn prepared_metadata_is_consumed_exactly_once() {
         let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new());
@@ -1049,7 +1036,7 @@ mod prepared_get_object_metadata_tests {
 
     #[test]
     #[serial_test::serial(body_cache_hook)]
-    fn inline_data_read_early_stop_defaults_return_exact_body() {
+    fn inline_data_read_early_stop_reader_returns_exact_body() {
         let runtime = tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()
@@ -1081,14 +1068,14 @@ mod prepared_get_object_metadata_tests {
 
                 temp_env::async_with_vars(
                     [
-                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", None::<&str>),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
-                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None::<&str>),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
-                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
                     ],
                     async {
-                        let slow_parity_disk = bounded_spare_disk_index(bucket, &object);
+                        let slow_initial_disk = bounded_slow_initial_disk_index(bucket, &object);
                         let barrier =
-                            rename_fanout_barrier::arm(&object, slow_parity_disk, rename_fanout_barrier::PHASE_READ_VERSION);
+                            rename_fanout_barrier::arm(&object, slow_initial_disk, rename_fanout_barrier::PHASE_READ_VERSION);
                         let calls = disk_call_counters::observe(&object);
                         let set_disks_for_read = Arc::clone(&set_disks);
                         let opts_for_read = opts.clone();
@@ -1101,10 +1088,10 @@ mod prepared_get_object_metadata_tests {
 
                         tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
                             .await
-                            .expect("default inline GET should pause a slow parity metadata read");
+                            .expect("bounded inline GET should pause a slow initial metadata read");
                         let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut open_reader)
                             .await
-                            .expect("default production inline GET should return before the paused parity metadata response")
+                            .expect("production inline GET should return before the paused metadata response")
                             .expect("inline GET reader task should not panic")
                             .expect("inline GET reader should open");
                         let object_size = reader.object_info.size;
@@ -1125,17 +1112,14 @@ mod prepared_get_object_metadata_tests {
 
         assert_eq!(object_size, payload.len() as i64);
         assert_eq!(restored, payload);
-        assert_eq!(
+        assert_eq!(calls_total, 4, "bounded production GET should schedule the initial quorum plus one spare");
-            calls_total, 4,
-            "default production inline GET should schedule the initial bounded quorum plus one hedge"
-        );
         assert_eq!(
             recorder.histogram_values(
                 "rustfs_io_get_object_metadata_fanout_scheduled",
                 &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
             ),
             vec![4.0],
-            "default production GET should record all scheduled metadata tasks"
+            "bounded production GET should record all scheduled metadata tasks"
         );
         assert_eq!(
             recorder.histogram_values(
@@ -1143,7 +1127,7 @@ mod prepared_get_object_metadata_tests {
                 &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
             ),
             vec![3.0],
-            "default production GET should record only observed metadata responses as completed"
+            "bounded production GET should record only observed metadata responses as completed"
         );
         assert_eq!(
             recorder.histogram_values(
@@ -1151,7 +1135,7 @@ mod prepared_get_object_metadata_tests {
                 &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
             ),
             vec![1.0],
-            "default production GET should record the aborted slow parity metadata task"
+            "bounded production GET should record the aborted slow metadata task"
         );
     }
 
@@ -1298,9 +1282,9 @@ mod prepared_get_object_metadata_tests {
 
                     temp_env::async_with_vars(
                         [
-                            ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", None::<&str>),
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
-                            ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None::<&str>),
+                            ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
-                            ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
                         ],
                         async {
                             let calls = disk_call_counters::observe(&object);
@@ -1702,95 +1686,6 @@ fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool {
     }
 }
 
-#[derive(Debug)]
-struct GetMetadataSlowtailFaultConfig {
-    delay: Duration,
-    disks: Arc<[usize]>,
-    bucket: Option<String>,
-    object_prefix: Option<String>,
-}
-
-#[derive(Clone, Debug)]
-struct GetMetadataSlowtailFaultRequest {
-    delay: Duration,
-    disks: Arc<[usize]>,
-}
-
-impl GetMetadataSlowtailFaultRequest {
-    fn delay_for_disk(&self, disk_index: usize) -> Option<Duration> {
-        self.disks.contains(&disk_index).then_some(self.delay)
-    }
-}
-
-fn parse_get_metadata_slowtail_fault_disks(raw: &str) -> Option<Vec<usize>> {
-    let mut disks = Vec::new();
-    for item in raw.split(',').map(str::trim).filter(|item| !item.is_empty()) {
-        let Ok(index) = item.parse::<usize>() else {
-            return None;
-        };
-        if !disks.contains(&index) {
-            disks.push(index);
-        }
-    }
-    (!disks.is_empty()).then_some(disks)
-}
-
-fn load_get_metadata_slowtail_fault_config() -> Option<GetMetadataSlowtailFaultConfig> {
-    let delay_ms = rustfs_utils::get_env_u64(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, 0);
-    if delay_ms == 0 {
-        return None;
-    }
-    let disks = parse_get_metadata_slowtail_fault_disks(&std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS).ok()?)?;
-    let bucket = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET)
-        .ok()
-        .filter(|value| !value.is_empty());
-    let object_prefix = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX)
-        .ok()
-        .filter(|value| !value.is_empty());
-    Some(GetMetadataSlowtailFaultConfig {
-        delay: Duration::from_millis(delay_ms),
-        disks: Arc::from(disks.into_boxed_slice()),
-        bucket,
-        object_prefix,
-    })
-}
-
-fn get_metadata_slowtail_fault_request(bucket: &str, object: &str, read_data: bool) -> Option<GetMetadataSlowtailFaultRequest> {
-    if !read_data {
-        return None;
-    }
-
-    #[cfg(test)]
-    let config = load_get_metadata_slowtail_fault_config();
-    #[cfg(test)]
-    let config = config.as_ref()?;
-    #[cfg(not(test))]
-    let config = ({
-        static CACHED: OnceLock<Option<GetMetadataSlowtailFaultConfig>> = OnceLock::new();
-        CACHED.get_or_init(load_get_metadata_slowtail_fault_config).as_ref()
-    })?;
-
-    if let Some(expected_bucket) = &config.bucket
-        && expected_bucket != bucket
-    {
-        return None;
-    }
-    if let Some(expected_prefix) = &config.object_prefix
-        && !object.starts_with(expected_prefix)
-    {
-        return None;
-    }
-    Some(GetMetadataSlowtailFaultRequest {
-        delay: config.delay,
-        disks: config.disks.clone(),
-    })
-}
-
-#[cfg(test)]
-fn get_metadata_slowtail_fault_delay(bucket: &str, object: &str, disk_index: usize, read_data: bool) -> Option<Duration> {
-    get_metadata_slowtail_fault_request(bucket, object, read_data)?.delay_for_disk(disk_index)
-}
-
 /// Check if multipart reads prefetch the next part's bitrot reader setup
 /// while the current part decodes (backlog#870).
 ///
@@ -1816,10 +1711,6 @@ fn is_multipart_reader_setup_prefetch_enabled() -> bool {
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
-)]
 fn get_metadata_early_stop_rollout_pct() -> u32 {
     static CACHED: OnceLock<u32> = OnceLock::new();
     *CACHED.get_or_init(|| {
@@ -1859,10 +1750,6 @@ fn should_use_codec_streaming(config: GetCodecStreamingConfig, bucket: &str, obj
 }
 
 /// Should this specific request use metadata early-stop?
-#[allow(
-    dead_code,
-    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
-)]
 pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool {
     let base = is_get_metadata_early_stop_enabled();
     let pct = get_metadata_early_stop_rollout_pct();
@@ -2296,7 +2183,6 @@ fn classify_get_codec_streaming_object_class(
     GetCodecStreamingObjectClass::PlainSinglePart
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn is_get_small_object_direct_memory_eligible_with_threshold(
     range: &Option<HTTPRangeSpec>,
     object_info: &ObjectInfo,
@@ -2902,7 +2788,6 @@ pub struct SetDisks {
     /// Stable namespace shared by every object lock created for this set.
     set_lock_namespace: Arc<str>,
     pub format: FormatV3,
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     disk_health_cache: Arc<RwLock<Vec<Option<DiskHealthEntry>>>>,
     get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
     get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState,
@@ -3178,13 +3063,11 @@ struct GetObjectMetadataCacheEntry {
 
 #[derive(Clone, Debug)]
 struct DiskHealthEntry {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     last_check: Instant,
     online: bool,
 }
 
 impl DiskHealthEntry {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn cached_value(&self) -> Option<bool> {
         if self.last_check.elapsed() <= DISK_HEALTH_CACHE_TTL {
             Some(self.online)
@@ -3778,7 +3661,6 @@ fn multipart_put_large_batch_min_size_bytes() -> usize {
     })
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
     if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
         SmallWritePath::Inline
@@ -3981,20 +3863,11 @@ fn inline_erasure_shard_file_offset(
 }
 
 fn collect_inline_data_shard_fileinfos_by_index<'a>(
-    parts_metadata: &'a [FileInfo],
-    fi: &FileInfo,
-    data_shards: usize,
-    disk_is_online: impl FnMut(usize) -> bool,
-) -> Option<Vec<&'a FileInfo>> {
-    collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, fi, data_shards, disk_is_online).ok()
-}
-
-fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
     parts_metadata: &'a [FileInfo],
     fi: &FileInfo,
     data_shards: usize,
     mut disk_is_online: impl FnMut(usize) -> bool,
-) -> std::result::Result<Vec<&'a FileInfo>, &'static str> {
+) -> Option<Vec<&'a FileInfo>> {
     let distribution = &fi.erasure.distribution;
     let mut data_files = vec![None; data_shards];
 
@@ -4002,35 +3875,27 @@ fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
         if !disk_is_online(disk_index) {
             continue;
         }
-        let Some(&block_index) = distribution.get(disk_index) else {
+        let block_index = *distribution.get(disk_index)?;
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
-        };
         if block_index == 0 || block_index > data_shards {
             continue;
         }
-        if file_info.name.is_empty() {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD);
-        }
         if file_info.erasure.index != block_index {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
+            continue;
         }
         if !file_info.has_valid_erasure_geometry() {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+            continue;
         }
         if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
+            continue;
         }
         if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
-            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_PAYLOAD);
+            continue;
         }
 
         data_files[block_index - 1] = Some(file_info);
     }
 
-    data_files
+    data_files.into_iter().collect()
-        .into_iter()
-        .collect::<Option<Vec<_>>>()
-        .ok_or(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD)
 }
 
 impl SetDisks {
@@ -4357,7 +4222,6 @@ fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &Obj
 ///
 /// Fail closed: when bucket metadata cannot be resolved the check stays on, so
 /// object-lock protection is never skipped because of a metadata lookup miss.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(crate) fn object_lock_delete_check_required(bucket_meta: Option<&crate::bucket::metadata::BucketMetadata>) -> bool {
     bucket_meta.is_none_or(|meta| meta.object_locking())
 }
@@ -4633,6 +4497,15 @@ impl Hash for ObjProps {
     }
 }
 
+#[derive(Default, Clone, Debug)]
+pub struct HealEntryResult {
+    pub bytes: usize,
+    pub success: bool,
+    pub skipped: bool,
+    pub entry_done: bool,
+    pub name: String,
+}
+
 fn is_object_dangling(
     meta_arr: &[FileInfo],
     errs: &[Option<DiskError>],
@@ -5409,7 +5282,6 @@ pub fn is_valid_storage_class(storage_class: &str) -> bool {
 }
 
 /// Returns true if the storage class is a cold storage tier that requires special handling
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_cold_storage_class(storage_class: &str) -> bool {
     matches!(
         storage_class,
@@ -5418,7 +5290,6 @@ pub fn is_cold_storage_class(storage_class: &str) -> bool {
 }
 
 /// Returns true if the storage class is an infrequent access tier
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_infrequent_access_class(storage_class: &str) -> bool {
     matches!(
         storage_class,

crates/ecstore/src/set_disk/ops/heal.rs

@@ -453,9 +453,7 @@ impl SetDisks {
             ..Default::default()
         };
 
-        // Bound, not `_`: this guard must live to the end of the scope. A bare
+        let write_lock_guard = if !opts.no_lock {
-        // `_` would drop it here and release the namespace write lock.
-        let _write_lock_guard = if !opts.no_lock {
             let ns_lock = self.new_ns_lock(bucket, object).await?;
             Some(
                 ns_lock
@@ -998,7 +996,7 @@ impl SetDisks {
                                                 readers.push(None);
                                                 continue;
                                             }
-                                            Err(_e) => {
+                                            Err(e) => {
                                                 readers.push(None);
                                                 continue;
                                             }
@@ -1547,9 +1545,6 @@ impl SetDisks {
 
         for candidate in candidates.iter_mut().filter(|candidate| candidate.local_payload) {
             for (disk_index, disk) in disks.iter().enumerate() {
-                // Only the #[cfg(test)] fault-injection branch below reads this.
-                #[cfg(not(test))]
-                let _ = disk_index;
                 let Some(disk) = disk else {
                     return Ok(DanglingDeleteSafety::UnsafeToDelete);
                 };
@@ -1721,10 +1716,6 @@ impl SetDisks {
         Ok((result, None))
     }
 
-    #[allow(
-        dead_code,
-        reason = "lock-taking wrapper over the live heal_object_dir_locked; only comments reference it (backlog#1823)"
-    )]
     #[tracing::instrument(level = "trace", skip(self), fields(bucket = %bucket, object = %object))]
     pub(in crate::set_disk) async fn heal_object_dir(
         &self,

crates/ecstore/src/set_disk/ops/locking.rs

@@ -66,7 +66,6 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
 }
 
 impl SetDisks {
-    #[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
     pub(in crate::set_disk) fn format_lock_error(&self, bucket: &str, object: &str, mode: &str, err: &LockResult) -> String {
         match err {
             LockResult::Timeout => {
@@ -80,7 +79,6 @@ impl SetDisks {
         }
     }
 
-    #[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
     pub(in crate::set_disk) fn format_lock_error_from_error(
         &self,
         bucket: &str,
@@ -145,7 +143,6 @@ impl SetDisks {
         disks
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn get_online_disks(&self) -> Vec<Option<DiskStore>> {
         let snapshot = self.drive_membership_snapshot().await;
         let mut disks = snapshot.strict_online_candidates().into_iter().map(Some).collect::<Vec<_>>();
@@ -156,10 +153,6 @@ impl SetDisks {
         disks
     }
 
-    #[allow(
-        dead_code,
-        reason = "local-only sibling of the test-covered get_online_disks; no caller in this port (backlog#1823)"
-    )]
     pub(in crate::set_disk) async fn get_online_local_disks(&self) -> Vec<Option<DiskStore>> {
         let snapshot = self.drive_membership_snapshot().await;
         let mut disks = snapshot
@@ -439,10 +432,6 @@ impl SetDisks {
         Ok((disk, fm))
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity healing-disk accessor with no caller in this port (backlog#1823)"
-    )]
     pub(in crate::set_disk) async fn get_online_disk_with_healing(
         &self,
         incl_healing: bool,
@@ -451,10 +440,6 @@ impl SetDisks {
         Ok((new_disks, healing > 0))
     }
 
-    #[allow(
-        dead_code,
-        reason = "reached only from get_online_disk_with_healing, itself uncalled in this port (backlog#1823)"
-    )]
     pub(in crate::set_disk) async fn get_online_disk_with_healing_and_info(
         &self,
         incl_healing: bool,

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -415,7 +415,6 @@ fn reduce_quorum_part_numbers(object_parts: Vec<Vec<String>>, read_quorum: usize
 /// never returned, but flips `is_truncated` to `true` and yields a
 /// `next_upload_id_marker` pointing at the last returned upload so the caller can
 /// resume paging.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn paginate_upload_page(remaining: &[MultipartInfo], max_uploads: usize) -> (Vec<MultipartInfo>, bool, Option<String>) {
     let is_truncated = remaining.len() > max_uploads;
     let page: Vec<MultipartInfo> = remaining.iter().take(max_uploads).cloned().collect();
@@ -558,7 +557,6 @@ impl SetDisks {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(super) async fn check_upload_id_exists(
         &self,
         bucket: &str,
@@ -1400,7 +1398,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         let mut count = max_parts;
 
-        for part in object_parts.iter() {
+        for (i, part) in object_parts.iter().enumerate() {
             if let Some(err) = &part.error {
                 warn!("list_object_parts part error: {:?}", &err);
             }
@@ -2043,8 +2041,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     && let Err(err) = checksum.add_part(&cs, ext_part.actual_size)
                 {
                     error!(
-                        "complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}, err={}",
+                        "complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}",
-                        p.part_num, bucket, object, err
+                        p.part_num, bucket, object
                     );
                     return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
                 }
@@ -2089,8 +2087,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 }
             } else if let Err(err) = wtcs.matches(&checksum_combined, uploaded_parts.len() as i32) {
                 error!(
-                    "complete_multipart_upload checksum matches failed want={}, got={}, err={}",
+                    "complete_multipart_upload checksum matches failed want={}, got={}",
-                    wtcs.encoded, checksum.encoded, err
+                    wtcs.encoded, checksum.encoded
                 );
                 return Err(Error::other(format!(
                     "complete_multipart_upload checksum matches failed want={}, got={}",

crates/ecstore/src/set_disk/ops/object.rs

@@ -3507,10 +3507,6 @@ struct TransitionUploadedSaveProbeState {
 }
 
 #[cfg(test)]
-#[allow(
-    dead_code,
-    reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-)]
 struct TransitionUploadedSaveProbe {
     state: Arc<TransitionUploadedSaveProbeState>,
 }
@@ -3521,10 +3517,6 @@ static TRANSITION_UPLOADED_SAVE_PROBE: std::sync::OnceLock<std::sync::Mutex<Opti
 
 #[cfg(test)]
 impl TransitionUploadedSaveProbe {
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn install(bucket: &str, object: &str) -> Self {
         let state = Arc::new(TransitionUploadedSaveProbeState {
             bucket: bucket.to_string(),
@@ -3541,10 +3533,6 @@ impl TransitionUploadedSaveProbe {
         Self { state }
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn attempts(&self) -> usize {
         self.state.attempts.load(std::sync::atomic::Ordering::Acquire)
     }
@@ -3750,10 +3738,6 @@ struct TransitionCommitBarrierState {
 }
 
 #[cfg(test)]
-#[allow(
-    dead_code,
-    reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-)]
 struct TransitionCommitBarrier {
     state: Arc<TransitionCommitBarrierState>,
 }
@@ -3764,26 +3748,14 @@ static TRANSITION_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Ar
 
 #[cfg(test)]
 impl TransitionCommitBarrier {
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn install_before_lock_lost_check(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::BeforeLockLost)
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn install(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::BeforeLeaseValidation)
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn install_after_lease_check(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::AfterLeaseValidation)
     }
@@ -3806,20 +3778,12 @@ impl TransitionCommitBarrier {
         Self { state }
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     async fn wait_until_paused(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
             .expect("transition should reach the deterministic commit barrier");
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn release(&self) {
         self.state.release.notify_one();
     }
@@ -5656,9 +5620,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         // TODO: Lifecycle
 
         let mut version_found = true;
-        // delete_object_version below derives its own majority quorum from the
+        let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
-        // disk array, so the object-derived quorum here is unused.
-        let (mut goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
         if let Some(err) = &gerr
             && goi.name.is_empty()
         {
@@ -6412,7 +6374,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
 
         for disk in disks.iter() {
-            if disk.is_some() {
+            if let Some(disk) = disk {
                 continue;
             }
             let _ = self
@@ -7543,7 +7505,7 @@ mod get_object_downstream_close_accounting_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks;
     use super::*;
     use crate::diagnostics::get::{
-        GET_METADATA_EARLY_STOP_REASON_NOT_FOUND, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT,
+        GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT,
         GetObjectFailureReason,
     };
     use crate::disk::RUSTFS_META_BUCKET;
@@ -7675,8 +7637,8 @@ mod get_object_downstream_close_accounting_tests {
             legacy_completed,
             internal_cancelled,
             legacy_cancelled,
-            internal_not_found_miss,
+            internal_unsafe_miss,
-            legacy_not_found_miss,
+            legacy_unsafe_miss,
             internal_saved,
             legacy_saved,
         ) = metrics::with_local_recorder(&recorder, || {
@@ -7752,7 +7714,7 @@ mod get_object_downstream_close_accounting_tests {
                         &[
                             ("path", GET_OBJECT_PATH_INTERNAL_META),
                             ("decision", "miss"),
-                            ("reason", GET_METADATA_EARLY_STOP_REASON_NOT_FOUND),
+                            ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
                         ],
                     ),
                     recorder.counter_value(
@@ -7760,7 +7722,7 @@ mod get_object_downstream_close_accounting_tests {
                         &[
                             ("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
                             ("decision", "miss"),
-                            ("reason", GET_METADATA_EARLY_STOP_REASON_NOT_FOUND),
+                            ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
                         ],
                     ),
                     recorder.histogram_values(
@@ -7811,21 +7773,21 @@ mod get_object_downstream_close_accounting_tests {
             "internal metadata lifecycle cancelled count must not leak into legacy_duplex"
         );
         assert_eq!(
-            internal_not_found_miss, 1,
+            internal_unsafe_miss, 1,
-            "internal metadata not-found early-stop miss must retain its path label"
+            "internal metadata unsafe early-stop miss must retain its path label"
         );
         assert_eq!(
-            legacy_not_found_miss, 0,
+            legacy_unsafe_miss, 0,
-            "internal metadata not-found early-stop miss must not leak into legacy_duplex"
+            "internal metadata unsafe early-stop miss must not leak into legacy_duplex"
         );
         assert_eq!(
             internal_saved,
             vec![0.0],
-            "internal metadata not-found miss must record zero saved responses on internal_meta"
+            "internal metadata unsafe miss must record zero saved responses on internal_meta"
         );
         assert!(
             legacy_saved.is_empty(),
-            "internal metadata not-found miss saved responses must not leak into legacy_duplex"
+            "internal metadata unsafe miss saved responses must not leak into legacy_duplex"
         );
     }
 }
@@ -10208,288 +10170,6 @@ mod transition_upload_integrity_tests {
         assert!(backend.contains(remote_object).await, "committed remote object should remain available");
     }
 
-    /// Compresses `plaintext` with the codec the PUT path uses, so the stored
-    /// bytes round-trip through the read path's decompressor.
-    async fn compress_for_storage(plaintext: &[u8]) -> Vec<u8> {
-        let mut reader = crate::io_support::rio::compression_reader(
-            Cursor::new(plaintext.to_vec()),
-            rustfs_utils::CompressionAlgorithm::default(),
-            false,
-        );
-        let mut compressed = Vec::new();
-        reader.read_to_end(&mut compressed).await.expect("plaintext should compress");
-        assert!(compressed.len() < plaintext.len(), "test payload must actually compress");
-        compressed
-    }
-
-    /// Writes a genuinely compressed object: stored data is `compressed`, and the
-    /// metadata marks it compressed with the plaintext length as its actual size,
-    /// exactly as the app-layer compress path records it.
-    async fn write_compressed_source(
-        set_disks: &Arc<SetDisks>,
-        disk_stores: &[DiskStore],
-        bucket: &str,
-        object: &str,
-        plaintext: &[u8],
-        compressed: &[u8],
-    ) -> ObjectInfo {
-        for disk in disk_stores {
-            disk.make_volume(bucket).await.expect("bucket volume should be created");
-        }
-        let mut user_defined = HashMap::new();
-        rustfs_utils::http::insert_str(
-            &mut user_defined,
-            rustfs_utils::http::SUFFIX_COMPRESSION,
-            crate::io_support::rio::compression_metadata_value(rustfs_utils::CompressionAlgorithm::default()),
-        );
-        rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
-        let stream = crate::io_support::rio::HashReader::from_stream(
-            Cursor::new(compressed.to_vec()),
-            compressed.len() as i64,
-            plaintext.len() as i64,
-            None,
-            None,
-            false,
-        )
-        .expect("hash reader over compressed bytes");
-        let mut reader = PutObjReader::new(stream);
-        set_disks
-            .put_object(
-                bucket,
-                object,
-                &mut reader,
-                &ObjectOptions {
-                    no_lock: true,
-                    user_defined,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("compressed object should be written")
-    }
-
-    async fn read_transitioned(
-        set_disks: &Arc<SetDisks>,
-        bucket: &str,
-        object: &str,
-        range: Option<HTTPRangeSpec>,
-        opts: &ObjectOptions,
-    ) -> (Vec<u8>, i64) {
-        let mut reader = set_disks
-            .get_object_reader(bucket, object, range, HeaderMap::new(), opts)
-            .await
-            .expect("transitioned object reader should open");
-        let published_size = reader.object_info.size;
-        let mut body = Vec::new();
-        reader
-            .stream
-            .read_to_end(&mut body)
-            .await
-            .expect("transitioned body should drain");
-        (body, published_size)
-    }
-
-    /// Transition uploads the object's STORED bytes, so a tiered read has to
-    /// apply the same transform an erasure read would. #6107 routed this path
-    /// through `ReadPlan` to stop serving an encrypted object's ciphertext;
-    /// compression rides the same plan, and nothing pinned it (backlog#1851).
-    /// Without the transform this GET returns the compressed bytes under the
-    /// compressed size — silent corruption for every client of a compressed
-    /// object that ILM has moved to a warm tier.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn transitioned_compressed_object_get_returns_plaintext() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "transitioned-compressed-get-bucket";
-        let object = "object.txt";
-        let plaintext = b"transitioned compressed objects must decompress on read ".repeat(20_000);
-        let compressed = compress_for_storage(&plaintext).await;
-        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
-
-        let opts = ObjectOptions {
-            no_lock: true,
-            ..Default::default()
-        };
-        let (local_body, local_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
-        assert_eq!(local_body, plaintext, "control: the pre-transition read must decompress");
-        assert_eq!(
-            local_size,
-            plaintext.len() as i64,
-            "control: the pre-transition read publishes the plaintext size"
-        );
-
-        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
-        set_disks
-            .transition_object(bucket, object, &transition_options(&original, tier_name))
-            .await
-            .expect("transition should commit");
-
-        let put_versions = backend.put_versions().await;
-        assert_eq!(put_versions.len(), 1, "transition should upload one remote candidate");
-        let remote_bytes = backend
-            .bytes(&put_versions[0].0)
-            .await
-            .expect("remote candidate should be stored");
-        assert_eq!(
-            remote_bytes, compressed,
-            "transition uploads the stored representation; the read side is what has to decode it"
-        );
-
-        let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
-        assert_eq!(body, plaintext, "a tiered read must return the object's content, not its stored bytes");
-        assert_eq!(
-            published_size,
-            plaintext.len() as i64,
-            "a tiered read must publish the plaintext size, not the compressed one"
-        );
-    }
-
-    /// A ranged tiered read is expressed in plaintext coordinates, so the plan
-    /// has to translate it into the remote copy's compressed extent and skip
-    /// into the decompressed stream — the same translation the erasure path does.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn transitioned_compressed_object_range_get_returns_plaintext_slice() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "transitioned-compressed-range-bucket";
-        let object = "object.txt";
-        let plaintext = b"ranged reads of transitioned compressed objects must land in plaintext ".repeat(20_000);
-        let compressed = compress_for_storage(&plaintext).await;
-        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
-
-        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
-        set_disks
-            .transition_object(bucket, object, &transition_options(&original, tier_name))
-            .await
-            .expect("transition should commit");
-
-        let opts = ObjectOptions {
-            no_lock: true,
-            ..Default::default()
-        };
-        // Deliberately past the compressed size, so a range still measured in
-        // stored coordinates could not produce this slice.
-        let start = compressed.len() as i64 + 4096;
-        let end = start + 511;
-        let range = HTTPRangeSpec {
-            is_suffix_length: false,
-            start,
-            end,
-        };
-        let (body, published_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
-
-        let expected = &plaintext[start as usize..=end as usize];
-        assert_eq!(body, expected, "a ranged tiered read must return that plaintext slice");
-        assert_eq!(published_size, expected.len() as i64, "a ranged tiered read publishes the slice length");
-    }
-
-    /// The restore copy-back re-writes the object under its original metadata,
-    /// which still says "compressed". It therefore has to keep receiving the
-    /// STORED bytes: `restore_request_active` holds it on the plan's `Plain`
-    /// branch, and decompressing there would write plaintext under compressed
-    /// metadata.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn restore_read_of_transitioned_compressed_object_keeps_stored_bytes() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "transitioned-compressed-restore-bucket";
-        let object = "object.txt";
-        let plaintext = b"restore copy-back must keep the stored representation intact ".repeat(20_000);
-        let compressed = compress_for_storage(&plaintext).await;
-        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
-
-        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
-        set_disks
-            .transition_object(bucket, object, &transition_options(&original, tier_name))
-            .await
-            .expect("transition should commit");
-
-        let oi = set_disks
-            .get_object_info(
-                bucket,
-                object,
-                &ObjectOptions {
-                    no_lock: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("transitioned metadata should resolve");
-        let restore_opts = ObjectOptions {
-            no_lock: true,
-            part_number: Some(1),
-            transition: TransitionOptions {
-                restore_request: s3s::dto::RestoreRequest {
-                    days: Some(1),
-                    ..Default::default()
-                },
-                ..Default::default()
-            },
-            ..Default::default()
-        };
-        let mut reader = get_transitioned_object_reader_with_tier_manager(
-            bucket,
-            object,
-            &None,
-            &HeaderMap::new(),
-            &oi,
-            &restore_opts,
-            &set_disks.ctx.tier_config_mgr(),
-            set_disks.ctx.object_encryption_resolver(),
-        )
-        .await
-        .expect("restore read of the tiered copy should open");
-        let published_size = reader.object_info.size;
-        let mut body = Vec::new();
-        reader.stream.read_to_end(&mut body).await.expect("restore body should drain");
-
-        assert_eq!(body, compressed, "a restore read must copy the stored bytes back verbatim");
-        assert_eq!(
-            published_size,
-            compressed.len() as i64,
-            "a restore read must keep publishing the stored size"
-        );
-    }
-
-    /// Plain objects must keep streaming the remote bytes through untouched:
-    /// their plan is `Plain`, so the tiered read stays byte-identical.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn transitioned_plain_object_get_is_unchanged() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "transitioned-plain-get-bucket";
-        let object = "object.bin";
-        let payload = b"plain transitioned objects must keep reading back byte-identical ".repeat(1024);
-        let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
-
-        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
-        set_disks
-            .transition_object(bucket, object, &transition_options(&original, tier_name))
-            .await
-            .expect("transition should commit");
-
-        let opts = ObjectOptions {
-            no_lock: true,
-            ..Default::default()
-        };
-        let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
-        assert_eq!(body, payload);
-        assert_eq!(published_size, payload.len() as i64);
-
-        let range = HTTPRangeSpec {
-            is_suffix_length: false,
-            start: 100,
-            end: 611,
-        };
-        let (ranged_body, ranged_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
-        assert_eq!(ranged_body, &payload[100..=611]);
-        assert_eq!(ranged_size, payload.len() as i64, "a plain ranged read keeps publishing the object size");
-    }
-
     async fn corrupt_beyond_read_quorum(
         temp_dirs: &[tempfile::TempDir],
         bucket: &str,

crates/ecstore/src/set_disk/read.rs

@@ -116,7 +116,6 @@ impl SetDisks {
             .then_some(GET_METADATA_CACHE_REASON_DIST_ERASURE)
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn cached_get_object_fileinfo(&self, bucket: &str, object: &str) -> Option<Arc<GetObjectMetadataCacheEntry>> {
         match self.lookup_cached_get_object_fileinfo(bucket, object).await {
             MetadataCacheLookup::Hit(entry) => Some(entry),
@@ -1827,7 +1826,6 @@ fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOp
         .then_some(GET_METADATA_CACHE_REASON_META_BUCKET)
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOptions, read_data: bool) -> bool {
     get_object_metadata_cache_request_bypass_reason(bucket, opts, read_data).is_none()
 }
@@ -3888,15 +3886,13 @@ mod tests {
                 assert!(metadata_early_stop_permitted(true, true, false, "", false, false));
                 // observe=false (non-observed fanout) also disables early-stop.
                 assert!(!metadata_early_stop_permitted(true, false, false, "", false, false));
-                // Whole/latest data-read metadata is now allowed by default;
+                assert!(!metadata_early_stop_permitted(true, true, true, "", false, false));
-                // the inline verifier still decides whether it can stop early.
-                assert!(metadata_early_stop_permitted(true, true, true, "", false, false));
             },
         );
     }
 
     #[test]
-    fn metadata_early_stop_allows_safe_data_reads_by_default() {
+    fn metadata_early_stop_keeps_data_reads_opt_in_by_default() {
         temp_env::with_vars(
             [
                 (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
@@ -3904,7 +3900,7 @@ mod tests {
                 (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, None),
             ],
             || {
-                assert!(should_allow_metadata_early_stop(true, "", false, false));
+                assert!(!should_allow_metadata_early_stop(true, "", false, false));
                 assert!(!should_allow_metadata_early_stop(true, "version-id", false, false));
                 assert!(should_allow_metadata_early_stop(false, "", false, false));
                 assert!(!should_allow_metadata_early_stop(false, "version-id", false, false));
@@ -3936,34 +3932,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn metadata_early_stop_bounded_fanout_defaults_to_enabled() {
-        temp_env::with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
-                (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, None),
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, None),
-            ],
-            || {
-                assert!(is_get_metadata_data_read_early_stop_enabled());
-                assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
-            },
-        );
-        temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false"))], || {
-            assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
-        });
-        temp_env::with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("false")),
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
-            ],
-            || {
-                assert!(!is_get_metadata_data_read_early_stop_enabled());
-                assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
-            },
-        );
-    }
-
     #[test]
     fn metadata_early_stop_rejects_healing_and_free_version_requests() {
         temp_env::with_vars(

crates/filemeta/src/filemeta/codec.rs

@@ -135,7 +135,10 @@ impl FileMeta {
         let i = buf.len() as u64;
 
         // check version, buf = buf[8..]
-        let (buf, _, _) = Self::check_xl2_v1(buf)?;
+        let (buf, _, _) = Self::check_xl2_v1(buf).map_err(|e| {
+            error!("failed to check XL2 v1 format: {}", e);
+            e
+        })?;
 
         if buf.len() < 5 {
             error!(

crates/heal/src/error.rs

@@ -82,8 +82,8 @@ impl Error {
     /// Whether a heal operation can be retried without changing its inputs.
     pub(crate) fn is_recoverable_heal(&self) -> bool {
         match self {
-            Error::TaskCancelled | Error::TaskTimeout => false,
+            Error::TaskCancelled => false,
-            Error::TransientSkip { .. } => true,
+            Error::TaskTimeout | Error::TransientSkip { .. } => true,
             Error::Storage(err) => {
                 err.is_quorum_error()
                     || matches!(
@@ -165,9 +165,4 @@ mod tests {
         assert!(Error::Storage(EcstoreError::DiskNotFound).is_recoverable_heal());
         assert!(Error::Storage(EcstoreError::VolumeNotFound).is_recoverable_heal());
     }
-
-    #[test]
-    fn task_timeout_is_terminal() {
-        assert!(!Error::TaskTimeout.is_recoverable_heal());
-    }
 }

crates/heal/src/heal/manager.rs

@@ -673,12 +673,6 @@ fn retry_request_for_result(task: &HealTask, result: &Result<()>) -> Option<(Hea
     Some((request, delay, error))
 }
 
-async fn retry_request_for_result_with_budget(task: &HealTask, result: &Result<()>) -> Option<(HealRequest, Duration, String)> {
-    let (_, delay, error) = retry_request_for_result(task, result)?;
-    let request = task.retry_request_with_remaining_timeout().await.ok()?;
-    Some((request, delay, error))
-}
-
 fn recoverable_heal_retry_delay(retry_attempt: u32) -> Duration {
     let retry_attempt = retry_attempt.clamp(1, 5);
     let delay = Duration::from_secs(2_u64.saturating_pow(retry_attempt));
@@ -696,7 +690,7 @@ pub struct HealConfig {
     pub max_concurrent_heals: usize,
     /// Maximum concurrent heal tasks allowed for a single erasure set
     pub max_concurrent_per_set: usize,
-    /// Aggregate task execution timeout across recoverable retries
+    /// Task timeout
     pub task_timeout: Duration,
     /// Queue size
     pub queue_size: usize,
@@ -3112,7 +3106,7 @@ impl HealManager {
                         "Heal scheduler task started"
                     );
                     let result = task.execute().await;
-                    let retry_request = retry_request_for_result_with_budget(task.as_ref(), &result).await;
+                    let retry_request = retry_request_for_result(task.as_ref(), &result);
                     match &result {
                         Ok(_) => {
                             debug!(
@@ -4545,25 +4539,6 @@ mod tests {
         assert!(retry_error.contains("Lock acquisition timeout"));
     }
 
-    #[tokio::test]
-    async fn retry_request_for_result_preserves_remaining_timeout_budget() {
-        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
-        let mut request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
-        request.options.timeout = Some(Duration::from_secs(60));
-        let task = HealTask::from_request(request, storage);
-        let result = task.execute().await;
-
-        let (retry_request, _, _) = retry_request_for_result_with_budget(&task, &result)
-            .await
-            .expect("read quorum failure should retain the unused timeout budget");
-        let remaining = retry_request
-            .options
-            .timeout
-            .expect("configured timeout should remain present");
-        assert!(remaining < Duration::from_secs(60));
-        assert!(remaining > Duration::from_secs(59));
-    }
-
     #[test]
     fn test_retry_request_for_incomplete_heal_rename() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -6079,7 +6054,7 @@ mod tests {
         process_manager_queue_once(&manager).await;
         let defaulted_status = tokio::time::timeout(Duration::from_secs(1), async {
             loop {
-                if let Ok(status @ HealTaskStatus::Timeout) = manager.get_task_status(&defaulted_id).await {
+                if let Ok(status @ HealTaskStatus::Retrying { .. }) = manager.get_task_status(&defaulted_id).await {
                     break status;
                 }
                 tokio::task::yield_now().await;
@@ -6087,8 +6062,23 @@ mod tests {
         })
         .await
         .expect("configured timeout should finish the task");
-        assert_eq!(defaulted_status, HealTaskStatus::Timeout);
+        assert!(matches!(defaulted_status, HealTaskStatus::Retrying { .. }));
-        assert!(manager.retrying_heals.lock().await.get(&defaulted_id).is_none());
+        assert_eq!(
+            manager
+                .retrying_heals
+                .lock()
+                .await
+                .get(&defaulted_id)
+                .expect("timed out task should retain its retry request")
+                .request
+                .options
+                .timeout,
+            Some(Duration::ZERO)
+        );
+        manager
+            .cancel_task(&defaulted_id)
+            .await
+            .expect("retrying timeout task should be cancelled");
 
         let mut explicit = bucket_request("explicit-timeout", HealPriority::Normal, HealRequestSource::Admin);
         explicit.options.timeout = Some(Duration::from_secs(60));

crates/heal/src/heal/task.rs

@@ -196,7 +196,7 @@ pub struct HealOptions {
     /// Whether to skip namespace locking
     #[serde(default)]
     pub no_lock: bool,
-    /// Aggregate execution timeout across recoverable manager retries
+    /// Timeout
     pub timeout: Option<Duration>,
     /// pool index
     pub pool_index: Option<usize>,
@@ -442,14 +442,6 @@ impl HealTask {
         }
     }
 
-    pub(crate) async fn retry_request_with_remaining_timeout(&self) -> Result<HealRequest> {
-        let mut request = self.retry_request();
-        if self.options.timeout.is_some() {
-            request.options.timeout = self.remaining_timeout().await?;
-        }
-        Ok(request)
-    }
-
     pub(crate) fn from_replacement_recovery_request(
         request: HealRequest,
         storage: Arc<dyn HealStorageAPI>,
@@ -2665,36 +2657,6 @@ mod tests {
 
     use super::super::storage_api::status::BucketInfo;
 
-    #[tokio::test]
-    async fn retry_request_carries_remaining_timeout_budget() {
-        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage::default());
-        let mut request = HealRequest::bucket("bucket".to_string());
-        request.options.timeout = Some(Duration::from_secs(100));
-        let task = HealTask::from_request(request, storage.clone());
-        *task.task_start_instant.write().await = Some(Instant::now() - Duration::from_secs(40));
-
-        let retry = task
-            .retry_request_with_remaining_timeout()
-            .await
-            .expect("first retry should retain the unused timeout budget");
-        let first_remaining = retry.options.timeout.expect("configured timeout should remain present");
-        assert!(first_remaining <= Duration::from_secs(60));
-        assert!(first_remaining > Duration::from_secs(59));
-
-        let retry_task = HealTask::from_request(retry, storage);
-        *retry_task.task_start_instant.write().await = Some(Instant::now() - Duration::from_secs(20));
-        let second_retry = retry_task
-            .retry_request_with_remaining_timeout()
-            .await
-            .expect("second retry should retain only the unused aggregate budget");
-        let second_remaining = second_retry
-            .options
-            .timeout
-            .expect("configured timeout should remain present");
-        assert!(second_remaining <= Duration::from_secs(40));
-        assert!(second_remaining > Duration::from_secs(39));
-    }
-
     #[test]
     fn format_result_requires_every_requested_target_to_be_ok() {
         let result = HealResultItem {

crates/io-metrics/src/lib.rs

@@ -487,21 +487,22 @@ pub fn record_get_object_completion(total_duration_secs: f64, response_size_byte
 
 /// Record the streaming strategy chosen for a GetObject response body.
 #[inline(always)]
-pub fn record_get_object_stream_strategy(strategy: &'static str, buffer_size_bytes: usize, response_size_bytes: i64) {
+pub fn record_get_object_stream_strategy(strategy: &str, buffer_size_bytes: usize, response_size_bytes: i64) {
     if !get_stage_metrics_enabled() {
         return;
     }
-    counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy).increment(1);
+    counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy.to_string()).increment(1);
-    histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy).record(usize_to_f64(buffer_size_bytes));
+    histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy.to_string())
-    histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy)
+        .record(usize_to_f64(buffer_size_bytes));
+    histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy.to_string())
         .record(i64_non_negative_to_f64(response_size_bytes));
 }
 
 /// Record the response-body handoff shape from a GetObject reader into the S3 streaming body.
 #[inline(always)]
 pub fn record_get_object_response_handoff(
-    strategy: &'static str,
+    strategy: &str,
-    buffer_source: &'static str,
+    buffer_source: &str,
     buffer_size_bytes: usize,
     response_size_bytes: i64,
     duration_secs: f64,
@@ -511,26 +512,26 @@ pub fn record_get_object_response_handoff(
     }
     counter!(
         "rustfs_io_get_object_response_handoff_total",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source
+        "buffer_source" => buffer_source.to_string()
     )
     .increment(1);
     histogram!(
         "rustfs_io_get_object_response_handoff_buffer_size_bytes",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source
+        "buffer_source" => buffer_source.to_string()
     )
     .record(usize_to_f64(buffer_size_bytes));
     histogram!(
         "rustfs_io_get_object_response_handoff_response_size_bytes",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source
+        "buffer_source" => buffer_source.to_string()
     )
     .record(i64_non_negative_to_f64(response_size_bytes));
     histogram!(
         "rustfs_io_get_object_response_handoff_duration_seconds",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source
+        "buffer_source" => buffer_source.to_string()
     )
     .record(duration_secs);
     record_get_object_response_handoff_duration("s3_handler", duration_secs);
@@ -538,18 +539,14 @@ pub fn record_get_object_response_handoff(
 
 /// Record ReaderStream capacity chosen for GetObject handoff.
 #[inline(always)]
-pub fn record_get_object_reader_stream_buffer_size(
+pub fn record_get_object_reader_stream_buffer_size(strategy: &str, buffer_source: &str, buffer_size_bytes: usize) {
-    strategy: &'static str,
-    buffer_source: &'static str,
-    buffer_size_bytes: usize,
-) {
     if !get_stage_metrics_enabled() {
         return;
     }
     histogram!(
         "rustfs_io_get_object_reader_stream_buffer_size_bytes",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source
+        "buffer_source" => buffer_source.to_string()
     )
     .record(usize_to_f64(buffer_size_bytes));
 }
@@ -557,8 +554,8 @@ pub fn record_get_object_reader_stream_buffer_size(
 /// Record ReaderStream poll outcomes for GetObject handoff attribution.
 #[inline(always)]
 pub fn record_get_object_reader_stream_poll(
-    strategy: &'static str,
+    strategy: &str,
-    buffer_source: &'static str,
+    buffer_source: &str,
     outcome: &'static str,
     remaining_before: usize,
     bytes: usize,
@@ -570,36 +567,36 @@ pub fn record_get_object_reader_stream_poll(
     let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
     counter!(
         "rustfs_io_get_object_reader_stream_poll_total",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source,
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .increment(1);
     counter!(
         "rustfs_io_get_object_reader_stream_poll_bytes_total",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source,
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .increment(bytes);
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_remaining_bytes",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source,
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .record(usize_to_f64(remaining_before));
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_bytes",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source,
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .record(usize_to_f64(bytes as usize));
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_duration_seconds",
-        "strategy" => strategy,
+        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source,
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .record(duration_secs);

crates/keystone/src/client.rs

@@ -31,10 +31,7 @@ pub struct KeystoneClient {
     admin_password: Option<String>,
     admin_project: Option<String>,
     admin_domain: String,
-    #[allow(
+    #[allow(dead_code)]
-        dead_code,
-        reason = "TLS verification flag parsed from config; the reqwest client is built before it is consulted, so nothing reads it back (backlog#1823)"
-    )]
     verify_ssl: bool,
     /// Request timeout applied to the underlying HTTP client.
     timeout: std::time::Duration,

crates/keystone/src/identity.rs

@@ -20,10 +20,7 @@ use tracing::{debug, info};
 
 /// Maps Keystone identities to RustFS concepts
 pub struct KeystoneIdentityMapper {
-    #[allow(
+    #[allow(dead_code)]
-        dead_code,
-        reason = "keeps the Keystone client alive for the mapper's lifetime; the mapping paths do not call through it yet (backlog#1823)"
-    )]
     client: Arc<KeystoneClient>,
     role_policy_map: HashMap<String, String>,
     enable_tenant_prefix: bool,

crates/kms/src/encryption/dek.rs

@@ -18,6 +18,8 @@
 //! data encryption keys using master keys. It abstracts the encryption
 //! operations so that different backends can share the same encryption logic.
 
+#![allow(dead_code)] // Trait methods may be used by implementations
+
 use crate::error::{KmsError, Result};
 use crate::persisted_observability::{BoundedUnknownFieldName, UnknownFieldSummary};
 use async_trait::async_trait;

crates/kms/tests/behavior_backup.rs

@@ -225,6 +225,8 @@ async fn nothing_readable_leaves_the_bundle_unwrapped() {
             "artifact {} carries the raw on-disk record",
             artifact.path
         );
+        // A cheap structural check too: an encrypted payload is not JSON.
+        assert_ne!(payload.first(), Some(&b'{'), "artifact {} looks like plaintext JSON", artifact.path);
     }
 
     // The manifest itself is not encrypted, so assert directly that it carries

crates/notify/src/rules/config.rs

@@ -40,10 +40,7 @@ impl RuleEvents for RuleView {
 #[derive(Debug)]
 struct CompiledRules {
     // Keep RulesMap (can be used later if you want to make more complex judgments during the snapshot reading phase)
-    #[allow(
+    #[allow(dead_code)]
-        dead_code,
-        reason = "speculative retention: the comment above keeps it for richer snapshot-time judgements that no code performs yet (backlog#1823)"
-    )]
     rules_map: RulesMap,
     // for RulesContainer::iter_rules
     rule_views: Vec<RuleView>,

crates/notify/src/rules/rules_map.rs

@@ -187,6 +187,7 @@ impl RulesMap {
     /// # Parameters
     /// * `event_name` - The EventName from which to remove the rule.
     /// * `pattern` - The pattern of the rule to be removed.
+    #[allow(dead_code)]
     pub fn remove_rule(&mut self, event_name: &EventName, pattern: &str) {
         let mut remove_event = false;
 
@@ -208,6 +209,7 @@ impl RulesMap {
     ///
     /// # Parameters
     /// * `event_names` - A slice of EventNames to be removed.
+    #[allow(dead_code)]
     pub fn remove_rules(&mut self, event_names: &[EventName]) {
         for event_name in event_names {
             self.map.remove(event_name);
@@ -221,6 +223,7 @@ impl RulesMap {
     /// * `event_name` - The EventName to update.
     /// * `pattern` - The pattern of the rule to be updated.
     /// * `target_id` - The TargetID to be added.
+    #[allow(dead_code)]
     pub fn update_rule(&mut self, event_name: EventName, pattern: String, target_id: TargetID) {
         self.map.entry(event_name).or_default().add(pattern, target_id);
         self.total_events_mask |= event_name.mask(); // Update only the relevant bitmask

crates/notify/src/rules/target_id_set.rs

@@ -18,6 +18,12 @@ use rustfs_targets::arn::TargetID;
 /// TargetIDSet - A collection representation of TargetID.
 pub type TargetIdSet = HashSet<TargetID>;
 
+/// Provides a Go-like method for TargetIdSet (can be implemented as trait if needed)
+#[allow(dead_code)]
+pub(crate) fn new_target_id_set(target_ids: Vec<TargetID>) -> TargetIdSet {
+    target_ids.into_iter().collect()
+}
+
 // HashSet has built-in clone, union, difference and other operations.
 // But the Go version of the method returns a new Set, and the HashSet method is usually iterator or modify itself.
 // If you need to exactly match Go's API style, you can add wrapper functions.

crates/obs/src/metrics/config.rs

@@ -17,6 +17,7 @@ use std::time::Duration;
 /// Environment variable key for the global default metrics interval (seconds).
 pub const ENV_DEFAULT_METRICS_INTERVAL: &str = "RUSTFS_METRICS_DEFAULT_INTERVAL_SEC";
 /// Default interval for metrics collection if not specified otherwise.
+#[allow(dead_code)]
 pub const DEFAULT_METRICS_INTERVAL: Duration = Duration::from_secs(60);
 
 /// Environment variable key for cluster metrics interval (seconds).

crates/obs/src/metrics/report.rs

@@ -145,18 +145,21 @@ impl PrometheusMetric {
     }
 
     #[inline]
+    #[allow(dead_code)]
     pub fn with_label(mut self, key: &'static str, value: impl Into<Cow<'static, str>>) -> Self {
         self.labels.push((key, value.into()));
         self
     }
 
     #[inline]
+    #[allow(dead_code)]
     pub fn with_label_owned(mut self, key: &'static str, value: String) -> Self {
         self.labels.push((key, Cow::Owned(value)));
         self
     }
 
     #[inline]
+    #[allow(dead_code)]
     pub fn with_labels(mut self, labels: Vec<(&'static str, Cow<'static, str>)>) -> Self {
         self.labels = labels;
         self

crates/obs/src/metrics/schema/entry/descriptor.rs

@@ -16,6 +16,7 @@ use crate::{MetricName, MetricNamespace, MetricSubsystem, MetricType};
 use std::collections::HashSet;
 
 /// MetricDescriptor - Metric descriptors
+#[allow(dead_code)]
 #[derive(Debug, Clone)]
 pub struct MetricDescriptor {
     pub name: MetricName,
@@ -51,6 +52,7 @@ impl MetricDescriptor {
     }
 
     /// Get the full metric name in Prometheus style: <namespace>_<subsystem>_<name>
+    #[allow(dead_code)]
     pub fn get_full_metric_name(&self) -> String {
         let namespace = self.namespace.as_str();
         let formatted_subsystem = self.subsystem.as_str();
@@ -59,6 +61,7 @@ impl MetricDescriptor {
     }
 
     /// check whether the label is in the label set
+    #[allow(dead_code)]
     pub fn has_label(&mut self, label: &str) -> bool {
         self.get_label_set().contains(label)
     }

crates/obs/src/metrics/schema/entry/metric_name.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 /// The metric name is the individual name of the metric
+#[allow(dead_code)]
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum MetricName {
     // The generic metric name
@@ -442,6 +443,7 @@ pub enum MetricName {
 }
 
 impl MetricName {
+    #[allow(dead_code)]
     pub fn as_str(&self) -> String {
         match self {
             Self::AuthTotal => "auth_total".to_string(),

crates/obs/src/metrics/schema/entry/metric_type.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 /// MetricType - Indicates the type of indicator
+#[allow(dead_code)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum MetricType {
     Counter,
@@ -22,6 +23,7 @@ pub enum MetricType {
 
 impl MetricType {
     /// convert the metric type to a string representation
+    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Counter => "counter",
@@ -32,6 +34,7 @@ impl MetricType {
 
     /// Convert the metric type to the Prometheus value type
     /// In a Rust implementation, this might return the corresponding Prometheus Rust client type
+    #[allow(dead_code)]
     pub fn as_prom(&self) -> &'static str {
         match self {
             Self::Counter => "counter.",

crates/obs/src/metrics/schema/entry/mod.rs

@@ -56,6 +56,7 @@ pub fn new_gauge_md(
 }
 
 /// create a new histogram indicator descriptor
+#[allow(dead_code)]
 pub fn new_histogram_md(
     name: impl Into<MetricName>,
     help: impl Into<String>,

crates/obs/src/metrics/schema/entry/namespace.rs

@@ -19,6 +19,7 @@ pub enum MetricNamespace {
 }
 
 impl MetricNamespace {
+    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::RustFS => "rustfs",

crates/obs/src/metrics/schema/entry/path_utils.rs

@@ -14,6 +14,7 @@
 
 /// Format the path to the metric name format
 /// Replace '/' and '-' with '_'
+#[allow(dead_code)]
 pub fn format_path_to_metric_name(path: &str) -> String {
     path.trim_start_matches('/').replace(['/', '-'], "_")
 }

crates/obs/src/metrics/schema/entry/subsystem.rs

@@ -102,6 +102,7 @@ impl MetricSubsystem {
     }
 
     /// Get the formatted metric name format string
+    #[allow(dead_code)]
     pub fn as_str(&self) -> String {
         format_path_to_metric_name(self.path())
     }
@@ -150,6 +151,7 @@ impl MetricSubsystem {
     }
 
     /// A convenient way to create custom subsystems directly
+    #[allow(dead_code)]
     pub fn new(path: impl Into<String>) -> Self {
         Self::Custom(path.into())
     }
@@ -174,6 +176,7 @@ impl std::fmt::Display for MetricSubsystem {
     }
 }
 
+#[allow(dead_code)]
 pub mod subsystems {
     use super::MetricSubsystem;
 

crates/obs/src/telemetry/rolling.rs

@@ -38,10 +38,7 @@ pub enum Rotation {
     Minutely,
     Hourly,
     Daily,
-    #[allow(
+    #[allow(dead_code)]
-        dead_code,
-        reason = "constructed only by this file's rolling-appender tests; the lib target cannot see them (backlog#1823)"
-    )]
     Never,
 }
 

crates/policy/src/policy/function.rs

@@ -219,6 +219,10 @@ impl PartialEq for Functions {
     }
 }
 
+#[derive(Clone, Serialize, Deserialize)]
+#[allow(dead_code)]
+pub struct Value;
+
 #[cfg(test)]
 mod tests {
     use crate::policy::Functions;

crates/policy/src/policy/utils/wildcard.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[allow(dead_code)]
 pub fn is_simple_match<P, N>(pattern: P, name: N) -> bool
 where
     P: AsRef<str>,
@@ -28,10 +29,7 @@ where
     inner_match(pattern, name, false)
 }
 
-#[allow(
+#[allow(dead_code)]
-    dead_code,
-    reason = "prefix-matcher asserted by this file's tests; no production caller yet (backlog#1823)"
-)]
 pub fn is_match_as_pattern_prefix<P, N>(pattern: P, text: N) -> bool
 where
     P: AsRef<str>,

crates/protocols/src/swift/account.rs

@@ -38,6 +38,7 @@ use std::collections::HashMap;
 /// - Account format is invalid
 /// - Credentials don't contain project_id
 /// - Account project_id doesn't match credentials project_id
+#[allow(dead_code)] // Used by Swift implementation
 pub fn validate_account_access(account: &str, credentials: &Credentials) -> SwiftResult<String> {
     // Extract project_id from account (strip "AUTH_" prefix)
     let account_project_id = account
@@ -69,6 +70,7 @@ pub fn validate_account_access(account: &str, credentials: &Credentials) -> Swif
 ///
 /// Admin users (with "admin" or "reseller_admin" roles) can perform
 /// cross-tenant operations and administrative tasks.
+#[allow(dead_code)] // Used by Swift implementation
 pub fn is_admin_user(credentials: &Credentials) -> bool {
     credentials
         .claims

crates/protocols/src/swift/container.rs

@@ -144,6 +144,7 @@ impl ContainerMapper {
     /// - S3 bucket name compatible (only uses [a-z0-9-])
     /// - Deterministic mapping (same input always produces same bucket name)
     /// - Fixed-length prefix (16 hex chars = 8 bytes)
+    #[allow(dead_code)] // Used in: create/delete container operations
     pub fn swift_to_s3_bucket(&self, container: &str, project_id: &str) -> String {
         if self.config.tenant_prefix_enabled {
             let hash = self.hash_project_id(project_id);
@@ -215,6 +216,7 @@ pub fn bucket_info_to_container(info: &BucketInfo, mapper: &ContainerMapper, pro
 /// 2. Lists all S3 buckets
 /// 3. Filters to buckets belonging to this tenant (using tenant prefix)
 /// 4. Converts BucketInfo to Swift Container format
+#[allow(dead_code)] // Used by handler: list containers
 pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftResult<Vec<Container>> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -277,6 +279,7 @@ pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftR
 /// - Returns 201 Created on success
 /// - Returns 202 Accepted if container already exists
 /// - Returns 400 Bad Request for invalid container names
+#[allow(dead_code)] // Used by handler
 pub async fn create_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<bool> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -345,6 +348,7 @@ fn validate_container_name(container: &str) -> SwiftResult<()> {
 }
 
 /// Container metadata for HEAD response
+#[allow(dead_code)] // TODO: Remove once Swift API integration is complete
 #[derive(Debug, Clone)]
 pub struct ContainerMetadata {
     /// Number of objects in container
@@ -407,6 +411,7 @@ pub(crate) async fn get_container_custom_metadata(
 /// - HEAD /v1/{account}/{container} returns container metadata
 /// - Returns 204 No Content on success with headers
 /// - Returns 404 Not Found if container doesn't exist
+#[allow(dead_code)] // Used by handler
 pub async fn get_container_metadata(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<ContainerMetadata> {
     let (bucket_name, bucket_info, custom_metadata) = get_container_metadata_base(account, container, credentials).await?;
 
@@ -443,6 +448,7 @@ pub async fn get_container_metadata(account: &str, container: &str, credentials:
 /// - The update is additive: items the request does not name keep their stored
 ///   value, and removal is explicit, via `X-Remove-Container-Meta-{name}` or an
 ///   empty value
+#[allow(dead_code)] // Used by handler
 pub async fn update_container_metadata(
     account: &str,
     container: &str,
@@ -514,6 +520,7 @@ pub async fn update_container_metadata(
 /// - Returns 204 No Content on success
 /// - Returns 404 Not Found if container doesn't exist
 /// - Returns 409 Conflict if container is not empty
+#[allow(dead_code)] // Used by handler
 pub async fn delete_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -596,6 +603,7 @@ pub async fn delete_container(account: &str, container: &str, credentials: &Cred
 /// - Account validation fails
 /// - Container doesn't exist
 /// - Storage layer errors occur
+#[allow(dead_code)] // Handler integration: GET container
 pub async fn list_objects(
     account: &str,
     container: &str,
@@ -698,6 +706,7 @@ pub async fn list_objects(
 /// Versioning configuration is stored as an S3 bucket tag:
 /// - Tag key: `swift-versions-location`
 /// - Tag value: archive container name
+#[allow(dead_code)] // Used by handler
 pub async fn enable_versioning(
     account: &str,
     container: &str,
@@ -786,6 +795,7 @@ pub async fn enable_versioning(
 /// * `account` - Account identifier
 /// * `container` - Container name to disable versioning on
 /// * `credentials` - Keystone credentials
+#[allow(dead_code)] // Used by handler
 pub async fn disable_versioning(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Validate account access
     let project_id = validate_account_access(account, credentials)?;
@@ -845,6 +855,7 @@ pub async fn disable_versioning(account: &str, container: &str, credentials: &Cr
 /// # Returns
 /// - Some(archive_container_name) if versioning is enabled
 /// - None if versioning is not enabled
+#[allow(dead_code)] // Used by handler and object.rs
 pub async fn get_versions_location(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<Option<String>> {
     // Validate account access
     let project_id = validate_account_access(account, credentials)?;
@@ -907,6 +918,7 @@ pub async fn get_versions_location(account: &str, container: &str, credentials:
 ///     &credentials
 /// ).await?;
 /// ```
+#[allow(dead_code)] // Used by handler
 pub async fn set_container_acl(
     account: &str,
     container: &str,
@@ -1010,6 +1022,7 @@ pub async fn set_container_acl(
 ///     println!("Container is publicly readable");
 /// }
 /// ```
+#[allow(dead_code)] // Used by handler
 pub async fn get_container_acl(
     account: &str,
     container: &str,
@@ -1070,6 +1083,7 @@ pub async fn get_container_acl(
 ///
 /// # Returns
 /// Ok(()) if ACLs were deleted successfully
+#[allow(dead_code)] // Used by handler
 pub async fn delete_container_acl(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Setting both ACLs to None removes them
     set_container_acl(account, container, None, None, credentials).await

crates/protocols/src/swift/errors.rs

@@ -20,6 +20,7 @@ use std::fmt;
 
 /// Swift-specific error type
 #[derive(Debug)]
+#[allow(dead_code)] // Error variants used by Swift implementation
 pub enum SwiftError {
     /// 400 Bad Request
     BadRequest(String),

crates/protocols/src/swift/object.rs

@@ -122,10 +122,12 @@ fn swift_user_metadata(headers: &HeaderMap) -> Option<HashMap<String, String>> {
 ///
 /// Handles URL encoding/decoding and path normalization for Swift object keys.
 /// Swift object names can contain any UTF-8 characters except null bytes.
+#[allow(dead_code)] // Used in: object operations
 pub struct ObjectKeyMapper;
 
 impl ObjectKeyMapper {
     /// Create a new object key mapper
+    #[allow(dead_code)] // Used in: object operations
     pub fn new() -> Self {
         Self
     }
@@ -138,6 +140,7 @@ impl ObjectKeyMapper {
     /// - Not contain null bytes
     /// - Not contain '..' path segments (directory traversal)
     /// - Not start with '/' (leading slash handled by routing)
+    #[allow(dead_code)] // Used in: object operations
     pub fn validate_object_name(object: &str) -> SwiftResult<()> {
         if object.is_empty() {
             return Err(SwiftError::BadRequest("Object name cannot be empty".to_string()));
@@ -180,6 +183,7 @@ impl ObjectKeyMapper {
     /// Example:
     /// - Swift: "photos/vacation/beach photo.jpg"
     /// - S3: "photos/vacation/beach photo.jpg"
+    #[allow(dead_code)] // Used in: object operations
     pub fn swift_to_s3_key(object: &str) -> SwiftResult<String> {
         Self::validate_object_name(object)?;
         Ok(object.to_string())
@@ -189,6 +193,7 @@ impl ObjectKeyMapper {
     ///
     /// This is essentially an identity transformation since we store
     /// Swift object names as-is in S3.
+    #[allow(dead_code)] // Used in: object operations
     pub fn s3_to_swift_name(key: &str) -> String {
         key.to_string()
     }
@@ -203,6 +208,7 @@ impl ObjectKeyMapper {
     /// - Object: "vacation/beach.jpg"
     /// - Bucket: "abc123:photos"
     /// - Key: "vacation/beach.jpg"
+    #[allow(dead_code)] // Used in: object operations
     pub fn build_s3_key(object: &str) -> SwiftResult<String> {
         Self::swift_to_s3_key(object)
     }
@@ -214,6 +220,7 @@ impl ObjectKeyMapper {
     ///
     /// Example URL: /v1/AUTH_abc/container/path%2Fto%2Ffile.txt
     /// Decoded: "path/to/file.txt"
+    #[allow(dead_code)] // Used in: object operations
     pub fn decode_object_from_url(encoded: &str) -> SwiftResult<String> {
         // Decode percent-encoding
         let decoded = urlencoding::decode(encoded).map_err(|e| SwiftError::BadRequest(format!("Invalid URL encoding: {}", e)))?;
@@ -226,6 +233,7 @@ impl ObjectKeyMapper {
     ///
     /// When constructing URLs (e.g., for redirect responses), we need to
     /// percent-encode object names.
+    #[allow(dead_code)] // Used in: object operations
     pub fn encode_object_for_url(object: &str) -> String {
         urlencoding::encode(object).to_string()
     }
@@ -233,6 +241,7 @@ impl ObjectKeyMapper {
     /// Check if object name represents a directory (pseudo-directory)
     ///
     /// In Swift, objects ending with '/' are treated as directory markers.
+    #[allow(dead_code)] // Used in: object operations
     pub fn is_directory_marker(object: &str) -> bool {
         object.ends_with('/')
     }
@@ -241,6 +250,7 @@ impl ObjectKeyMapper {
     ///
     /// Removes redundant slashes and normalizes the path while preserving
     /// trailing slashes for directory markers.
+    #[allow(dead_code)] // Used in: object operations
     pub fn normalize_path(object: &str) -> String {
         // Split by '/', filter out empty segments (except if it's the end)
         let has_trailing_slash = object.ends_with('/');
@@ -314,6 +324,7 @@ fn sanitize_storage_error<E: std::fmt::Display>(operation: &str, error: E) -> Sw
 /// # Returns
 /// * `Ok(etag)` - Object ETag on success
 /// * `Err(SwiftError)` - Error if validation fails or upload fails
+#[allow(dead_code)] // Handler integration: PUT object
 pub async fn put_object<R>(
     account: &str,
     container: &str,
@@ -434,6 +445,7 @@ where
 ///
 /// Similar to put_object, but allows directly specifying metadata instead of extracting from headers.
 /// This is used internally for storing SLO manifests and marker objects.
+#[allow(dead_code)] // Used by SLO implementation
 pub async fn put_object_with_metadata<R>(
     account: &str,
     container: &str,
@@ -537,6 +549,7 @@ where
 /// - `bytes=1000-1999` - Bytes 1000-1999
 /// - `bytes=1000-` - From byte 1000 to end
 /// - `bytes=-500` - Last 500 bytes
+#[allow(dead_code)] // Handler integration: GET object
 pub async fn get_object(
     account: &str,
     container: &str,
@@ -595,6 +608,7 @@ pub async fn get_object(
 /// # Returns
 /// * `Ok(object_info)` - Object metadata (ObjectInfo)
 /// * `Err(SwiftError)` - Error if validation fails or object not found
+#[allow(dead_code)] // Handler integration: HEAD object
 pub async fn head_object(
     account: &str,
     container: &str,
@@ -657,6 +671,7 @@ pub async fn head_object(
 /// # Returns
 /// * `Ok(())` - Object deleted successfully (or didn't exist)
 /// * `Err(SwiftError)` - Error if validation fails or deletion fails
+#[allow(dead_code)] // Handler integration: DELETE object
 pub async fn delete_object(account: &str, container: &str, object: &str, credentials: &Credentials) -> SwiftResult<()> {
     // 1. Validate account access and get project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -717,6 +732,7 @@ pub async fn delete_object(account: &str, container: &str, object: &str, credent
 /// # Returns
 /// * `Ok(())` - Metadata updated successfully
 /// * `Err(SwiftError)` - Error if validation fails, object not found, or update fails
+#[allow(dead_code)] // Handler integration: POST object
 pub async fn update_object_metadata(
     account: &str,
     container: &str,
@@ -830,6 +846,7 @@ pub async fn update_object_metadata(
 /// # Handler Integration Note
 /// The current handler architecture needs to be updated to pass headers through
 /// to support COPY method and X-Copy-From header detection. See handler.rs for details.
+#[allow(dead_code)] // Handler integration: COPY object
 #[allow(clippy::too_many_arguments)] // Necessary for full copy functionality
 pub async fn copy_object(
     src_account: &str,
@@ -962,6 +979,7 @@ pub async fn copy_object(
 /// assert_eq!(container, "my-container");
 /// assert_eq!(object, "path/to/file.txt");
 /// ```
+#[allow(dead_code)] // Handler integration: COPY method
 pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, String)> {
     let destination = destination.trim_start_matches('/');
     let parts: Vec<&str> = destination.splitn(2, '/').collect();
@@ -995,6 +1013,7 @@ pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, Strin
 /// # Returns
 /// * `Ok((container, object))` - Parsed container and object names
 /// * `Err(SwiftError)` - Error if format is invalid
+#[allow(dead_code)] // Handler integration: X-Copy-From
 pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)> {
     // Same parsing logic as Destination header
     parse_destination_header(copy_from)
@@ -1023,6 +1042,7 @@ pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)>
 /// assert_eq!(range.start, 0);
 /// assert_eq!(range.end, 1023);
 /// ```
+#[allow(dead_code)] // Handler integration: Range header
 pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
     if !range_str.starts_with("bytes=") {
         return Err(SwiftError::BadRequest("Range header must start with 'bytes='".to_string()));
@@ -1104,6 +1124,7 @@ pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
 /// let header = format_content_range(0, 1023, 5000);
 /// assert_eq!(header, "bytes 0-1023/5000");
 /// ```
+#[allow(dead_code)] // Handler integration: Range header
 pub fn format_content_range(start: i64, end: i64, total: i64) -> String {
     format!("bytes {}-{}/{}", start, end, total)
 }

crates/protocols/src/swift/router.rs

@@ -50,6 +50,7 @@ pub enum SwiftRoute {
 
 impl SwiftRoute {
     /// Get the account identifier from the route
+    #[allow(dead_code)] // Public API for future use
     pub fn account(&self) -> &str {
         match self {
             SwiftRoute::Account { account, .. } => account,
@@ -59,6 +60,7 @@ impl SwiftRoute {
     }
 
     /// Extract project_id from account string (removes AUTH_ prefix)
+    #[allow(dead_code)] // Public API for future use
     pub fn project_id(&self) -> Option<&str> {
         let account = self.account();
         ACCOUNT_PATTERN