chore(release): prepare 1.0.3

fix(scanner): clear transient cycle recovery on follower lock contention (1.0.2 backport) (#8596 )
fix(scanner): clear transient cycle recovery on follower lock contention (#8566) (cherry picked from commit 606d8a3a8a) Co-authored-by: Ramakrishna Chilaka <49393831+RamakrishnaChilaka@users.noreply.github.com>
2026-10-27 22:37:13 +00:00 · 2026-10-17 15:57:35 +08:00 · 2026-10-16 04:05:23 +08:00 · 2026-10-15 18:11:29 +00:00 · 2026-10-15 18:00:45 +00:00 · 2026-10-15 22:47:37 +08:00
23 changed files with 2224 additions and 317 deletions
@@ -4030,7 +4030,7 @@ checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"

 [[package]]
 name = "e2e_test"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "anyhow",
 "async-compression",
@@ -9522,7 +9522,7 @@ dependencies = [

 [[package]]
 name = "rustfs"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "aes-gcm",
 "anyhow",
@@ -9681,7 +9681,7 @@ dependencies = [

 [[package]]
 name = "rustfs-audit"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "const-str",
 "futures",
@@ -9703,7 +9703,7 @@ dependencies = [

 [[package]]
 name = "rustfs-checksums"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "base64-simd",
 "bytes",
@@ -9719,7 +9719,7 @@ dependencies = [

 [[package]]
 name = "rustfs-common"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "arc-swap",
 "hotpath",
@@ -9734,7 +9734,7 @@ dependencies = [

 [[package]]
 name = "rustfs-concurrency"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hotpath",
 "insta",
@@ -9747,7 +9747,7 @@ dependencies = [

 [[package]]
 name = "rustfs-config"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "const-str",
 "hotpath",
@@ -9757,7 +9757,7 @@ dependencies = [

 [[package]]
 name = "rustfs-connect-symbol-catalog"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hex-simd",
 "object 0.40.0",
@@ -9769,7 +9769,7 @@ dependencies = [

 [[package]]
 name = "rustfs-credentials"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "base64-simd",
 "hmac 0.13.0",
@@ -9783,7 +9783,7 @@ dependencies = [

 [[package]]
 name = "rustfs-crypto"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "aes-gcm",
 "argon2",
@@ -9804,7 +9804,7 @@ dependencies = [

 [[package]]
 name = "rustfs-data-usage"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hotpath",
 "rmp-serde",
@@ -9814,7 +9814,7 @@ dependencies = [

 [[package]]
 name = "rustfs-ecstore"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "arc-swap",
 "async-channel",
@@ -9947,7 +9947,7 @@ dependencies = [

 [[package]]
 name = "rustfs-extension-schema"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hotpath",
 "serde",
@@ -9957,7 +9957,7 @@ dependencies = [

 [[package]]
 name = "rustfs-filemeta"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "arc-swap",
 "base64-simd",
@@ -9986,7 +9986,7 @@ dependencies = [

 [[package]]
 name = "rustfs-heal"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-trait",
 "base64-simd",
@@ -10028,7 +10028,7 @@ dependencies = [

 [[package]]
 name = "rustfs-heal-contracts"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "serde",
 "serde_json",
@@ -10038,7 +10038,7 @@ dependencies = [

 [[package]]
 name = "rustfs-iam"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "arc-swap",
 "async-trait",
@@ -10088,7 +10088,7 @@ dependencies = [

 [[package]]
 name = "rustfs-io-core"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "bytes",
 "hotpath",
@@ -10100,7 +10100,7 @@ dependencies = [

 [[package]]
 name = "rustfs-io-metrics"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "criterion",
 "hotpath",
@@ -10164,7 +10164,7 @@ dependencies = [

 [[package]]
 name = "rustfs-keystone"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "bytes",
 "futures",
@@ -10191,7 +10191,7 @@ dependencies = [

 [[package]]
 name = "rustfs-kms"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "aes-gcm",
 "anyhow",
@@ -10241,14 +10241,14 @@ dependencies = [

 [[package]]
 name = "rustfs-license"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "thiserror 2.0.21",
 ]

 [[package]]
 name = "rustfs-lifecycle"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-trait",
 "hotpath",
@@ -10270,7 +10270,7 @@ dependencies = [

 [[package]]
 name = "rustfs-lock"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-trait",
 "compact_str",
@@ -10293,7 +10293,7 @@ dependencies = [

 [[package]]
 name = "rustfs-log-analyzer"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "chrono",
 "flate2",
@@ -10312,7 +10312,7 @@ dependencies = [

 [[package]]
 name = "rustfs-madmin"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hotpath",
 "http 1.5.0",
@@ -10350,7 +10350,7 @@ dependencies = [

 [[package]]
 name = "rustfs-notify"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "arc-swap",
 "async-trait",
@@ -10385,7 +10385,7 @@ dependencies = [

 [[package]]
 name = "rustfs-object-capacity"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "criterion",
 "futures",
@@ -10404,7 +10404,7 @@ dependencies = [

 [[package]]
 name = "rustfs-object-data-cache"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "bytes",
 "criterion",
@@ -10421,7 +10421,7 @@ dependencies = [

 [[package]]
 name = "rustfs-obs"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "chrono",
 "crossbeam-channel",
@@ -10484,7 +10484,7 @@ dependencies = [

 [[package]]
 name = "rustfs-policy"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-trait",
 "base64-simd",
@@ -10515,7 +10515,7 @@ dependencies = [

 [[package]]
 name = "rustfs-protocols"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-compression",
 "async-trait",
@@ -10577,7 +10577,7 @@ dependencies = [

 [[package]]
 name = "rustfs-protos"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "flatbuffers",
 "hotpath",
@@ -10602,7 +10602,7 @@ dependencies = [

 [[package]]
 name = "rustfs-replication"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "byteorder",
 "bytes",
@@ -10620,7 +10620,7 @@ dependencies = [

 [[package]]
 name = "rustfs-rio"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "aes-gcm",
 "arc-swap",
@@ -10664,7 +10664,7 @@ dependencies = [

 [[package]]
 name = "rustfs-rio-v2"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "aes-gcm",
 "bytes",
@@ -10687,7 +10687,7 @@ dependencies = [

 [[package]]
 name = "rustfs-s3-client"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "base64-simd",
 "bytes",
@@ -10731,7 +10731,7 @@ dependencies = [

 [[package]]
 name = "rustfs-s3-ops"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hotpath",
 "rustfs-s3-types",
@@ -10739,7 +10739,7 @@ dependencies = [

 [[package]]
 name = "rustfs-s3-types"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hotpath",
 "serde",
@@ -10748,7 +10748,7 @@ dependencies = [

 [[package]]
 name = "rustfs-s3select-api"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "arc-swap",
 "async-compression",
@@ -10784,7 +10784,7 @@ dependencies = [

 [[package]]
 name = "rustfs-s3select-query"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-trait",
 "bytes",
@@ -10804,7 +10804,7 @@ dependencies = [

 [[package]]
 name = "rustfs-scanner"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-trait",
 "bytes",
@@ -10851,7 +10851,7 @@ dependencies = [

 [[package]]
 name = "rustfs-scanner-metrics"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "chrono",
 "jiff",
@@ -10866,7 +10866,7 @@ dependencies = [

 [[package]]
 name = "rustfs-security-governance"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hotpath",
 "thiserror 2.0.21",
@@ -10874,7 +10874,7 @@ dependencies = [

 [[package]]
 name = "rustfs-signer"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "base64-simd",
 "bytes",
@@ -10892,7 +10892,7 @@ dependencies = [

 [[package]]
 name = "rustfs-storage-api"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-trait",
 "hotpath",
@@ -10907,7 +10907,7 @@ dependencies = [

 [[package]]
 name = "rustfs-targets"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "arc-swap",
 "async-nats",
@@ -10961,7 +10961,7 @@ dependencies = [

 [[package]]
 name = "rustfs-test-utils"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "hotpath",
 "rustfs-data-usage",
@@ -10977,7 +10977,7 @@ dependencies = [

 [[package]]
 name = "rustfs-tls-runtime"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "arc-swap",
 "hotpath",
@@ -11016,7 +11016,7 @@ dependencies = [

 [[package]]
 name = "rustfs-trusted-proxies"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-trait",
 "axum",
@@ -11053,7 +11053,7 @@ dependencies = [

 [[package]]
 name = "rustfs-utils"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "base64-simd",
 "blake2",
@@ -11095,7 +11095,7 @@ dependencies = [

 [[package]]
 name = "rustfs-zip"
-version = "1.0.2"
+version = "1.0.3"
 dependencies = [
 "async-compression",
 "futures",
@@ -74,7 +74,7 @@ edition = "2024"
 license = "Apache-2.0"
 repository = "https://github.com/rustfs/rustfs"
 rust-version = "1.99.0"
-version = "1.0.2"
+version = "1.0.3"
 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"]
@@ -91,56 +91,56 @@ redundant_clone = "warn"

 [workspace.dependencies]
 # RustFS Internal Crates
-rustfs = { path = "./rustfs", version = "1.0.2" }
-rustfs-heal = { path = "crates/heal", version = "1.0.2" }
-rustfs-heal-contracts = { path = "crates/heal-contracts", version = "1.0.2" }
-rustfs-scanner-metrics = { path = "crates/scanner-metrics", version = "1.0.2" }
-rustfs-audit = { path = "crates/audit", version = "1.0.2" }
-rustfs-checksums = { path = "crates/checksums", version = "1.0.2" }
-rustfs-common = { path = "crates/common", version = "1.0.2" }
-rustfs-data-usage = { path = "crates/data-usage", version = "1.0.2" }
-rustfs-config = { path = "./crates/config", version = "1.0.2" }
-rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.2" }
-rustfs-credentials = { path = "crates/credentials", version = "1.0.2" }
-rustfs-crypto = { path = "crates/crypto", version = "1.0.2" }
-rustfs-ecstore = { path = "crates/ecstore", version = "1.0.2" }
-rustfs-filemeta = { path = "crates/filemeta", version = "1.0.2" }
-rustfs-iam = { path = "crates/iam", version = "1.0.2" }
-rustfs-keystone = { path = "crates/keystone", version = "1.0.2" }
-rustfs-license = { path = "crates/license", version = "1.0.2" }
-rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.2" }
-rustfs-kms = { path = "crates/kms", version = "1.0.2" }
-rustfs-lock = { path = "crates/lock", version = "1.0.2" }
-rustfs-madmin = { path = "crates/madmin", version = "1.0.2" }
-rustfs-notify = { path = "crates/notify", version = "1.0.2" }
-rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.2" }
-rustfs-io-core = { path = "crates/io-core", version = "1.0.2" }
-rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.2" }
-rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.2", default-features = false }
-rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.2" }
-rustfs-obs = { path = "crates/obs", version = "1.0.2" }
-rustfs-policy = { path = "crates/policy", version = "1.0.2" }
-rustfs-protos = { path = "crates/protos", version = "1.0.2" }
-rustfs-protocols = { path = "crates/protocols", version = "1.0.2" }
-rustfs-replication = { path = "crates/replication", version = "1.0.2" }
-rustfs-rio = { path = "crates/rio", version = "1.0.2" }
-rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.2" }
-rustfs-s3-client = { path = "crates/s3-client", version = "1.0.2" }
-rustfs-s3-types = { path = "crates/s3-types", version = "1.0.2" }
-rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.2" }
-rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.2" }
-rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.2" }
-rustfs-scanner = { path = "crates/scanner", version = "1.0.2" }
-rustfs-security-governance = { path = "crates/security-governance", version = "1.0.2" }
-rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.2" }
-rustfs-signer = { path = "crates/signer", version = "1.0.2" }
-rustfs-storage-api = { path = "crates/storage-api", version = "1.0.2" }
-rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.2" }
-rustfs-targets = { path = "crates/targets", version = "1.0.2" }
-rustfs-test-utils = { path = "crates/test-utils", version = "1.0.2" }
-rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.2" }
-rustfs-utils = { path = "crates/utils", version = "1.0.2" }
-rustfs-zip = { path = "./crates/zip", version = "1.0.2" }
+rustfs = { path = "./rustfs", version = "1.0.3" }
+rustfs-heal = { path = "crates/heal", version = "1.0.3" }
+rustfs-heal-contracts = { path = "crates/heal-contracts", version = "1.0.3" }
+rustfs-scanner-metrics = { path = "crates/scanner-metrics", version = "1.0.3" }
+rustfs-audit = { path = "crates/audit", version = "1.0.3" }
+rustfs-checksums = { path = "crates/checksums", version = "1.0.3" }
+rustfs-common = { path = "crates/common", version = "1.0.3" }
+rustfs-data-usage = { path = "crates/data-usage", version = "1.0.3" }
+rustfs-config = { path = "./crates/config", version = "1.0.3" }
+rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.3" }
+rustfs-credentials = { path = "crates/credentials", version = "1.0.3" }
+rustfs-crypto = { path = "crates/crypto", version = "1.0.3" }
+rustfs-ecstore = { path = "crates/ecstore", version = "1.0.3" }
+rustfs-filemeta = { path = "crates/filemeta", version = "1.0.3" }
+rustfs-iam = { path = "crates/iam", version = "1.0.3" }
+rustfs-keystone = { path = "crates/keystone", version = "1.0.3" }
+rustfs-license = { path = "crates/license", version = "1.0.3" }
+rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.3" }
+rustfs-kms = { path = "crates/kms", version = "1.0.3" }
+rustfs-lock = { path = "crates/lock", version = "1.0.3" }
+rustfs-madmin = { path = "crates/madmin", version = "1.0.3" }
+rustfs-notify = { path = "crates/notify", version = "1.0.3" }
+rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.3" }
+rustfs-io-core = { path = "crates/io-core", version = "1.0.3" }
+rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.3" }
+rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.3", default-features = false }
+rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.3" }
+rustfs-obs = { path = "crates/obs", version = "1.0.3" }
+rustfs-policy = { path = "crates/policy", version = "1.0.3" }
+rustfs-protos = { path = "crates/protos", version = "1.0.3" }
+rustfs-protocols = { path = "crates/protocols", version = "1.0.3" }
+rustfs-replication = { path = "crates/replication", version = "1.0.3" }
+rustfs-rio = { path = "crates/rio", version = "1.0.3" }
+rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.3" }
+rustfs-s3-client = { path = "crates/s3-client", version = "1.0.3" }
+rustfs-s3-types = { path = "crates/s3-types", version = "1.0.3" }
+rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.3" }
+rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.3" }
+rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.3" }
+rustfs-scanner = { path = "crates/scanner", version = "1.0.3" }
+rustfs-security-governance = { path = "crates/security-governance", version = "1.0.3" }
+rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.3" }
+rustfs-signer = { path = "crates/signer", version = "1.0.3" }
+rustfs-storage-api = { path = "crates/storage-api", version = "1.0.3" }
+rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.3" }
+rustfs-targets = { path = "crates/targets", version = "1.0.3" }
+rustfs-test-utils = { path = "crates/test-utils", version = "1.0.3" }
+rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.3" }
+rustfs-utils = { path = "crates/utils", version = "1.0.3" }
+rustfs-zip = { path = "./crates/zip", version = "1.0.3" }

 # Async Runtime and Networking
 async-channel = "2.5.0"
@@ -860,10 +860,12 @@ pub(crate) async fn start_rebalance(cluster: &RustFSTestClusterEnvironment) -> T
        if status.is_success() {
            break response;
        }
-        if status == StatusCode::INTERNAL_SERVER_ERROR
-            && response.contains("requires a live fleet capability proof")
-            && Instant::now() < deadline
-        {
+        // The init stage answers 503 + Retry-After while the cluster capability
+        // proof is still missing; older binaries surfaced the same condition as
+        // a 500 carrying the proof marker.
+        let waiting_for_readiness = status == StatusCode::SERVICE_UNAVAILABLE
+            || (status == StatusCode::INTERNAL_SERVER_ERROR && response.contains("requires a live fleet capability proof"));
+        if waiting_for_readiness && Instant::now() < deadline {
            sleep(Duration::from_secs(1)).await;
            continue;
        }
@@ -65,9 +65,9 @@ use super::replication_target_boundary::{
    SsecPassthroughGate, TargetClient, VersionIdentityCapability, is_replication_target_offline_error,
    replication_action_for_target_head, replication_complete_multipart_options, replication_delete_marker_purge_remove_options,
    replication_delete_remove_options, replication_force_delete_remove_options, replication_object_is_ssec_encrypted,
-    replication_put_object_header_size, replication_put_object_options, replication_target_head_is_newer_null_version,
-    resolve_read_api_version_id, ssec_passthrough_evidence_present, ssec_passthrough_gate, version_identity_capability_from_put,
-    version_identity_drifted,
+    replication_put_object_header_size, replication_put_object_options, replication_source_etag_is_comparable,
+    replication_target_head_is_newer_null_version, resolve_read_api_version_id, ssec_passthrough_evidence_present,
+    ssec_passthrough_gate, version_identity_capability_from_put, version_identity_drifted,
 };
 use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
@@ -349,15 +349,22 @@ fn replica_etag_verification_enabled() -> bool {
 /// target that stores a transformed payload (e.g. undecoded `aws-chunked`
 /// frames, #6853) returns the ETag of what it actually wrote. Reporting
 /// COMPLETED over such a replica is silent corruption, so a decidable
-/// mismatch fails the replication instead. An SSE-C ciphertext passthrough
-/// transfer is exempt: the wire bytes are ciphertext while the source ETag is
-/// the plaintext MD5, and that path has its own HEAD-back audit.
+/// mismatch fails the replication instead. Encrypted sources are exempt: an
+/// SSE-C ciphertext passthrough sends ciphertext while the source ETag is the
+/// plaintext MD5 (that path has its own HEAD-back audit), and a managed-SSE
+/// (SSE-S3 / SSE-KMS) source stores a keyed fingerprint, or a legacy plaintext
+/// MD5, that the ETag of the target's re-encrypted replica can neither confirm
+/// nor refute. The checksum header on the replication PUT stays the integrity
+/// check for those objects.
 fn verify_single_part_replica(
    object_info: &ObjectInfo,
    response: &RemotePutObjectResponse,
    ciphertext_passthrough: bool,
 ) -> std::result::Result<(), std::io::Error> {
-    if ciphertext_passthrough || !replica_etag_verification_enabled() {
+    if ciphertext_passthrough
+        || !replication_source_etag_is_comparable(&object_info.user_defined)
+        || !replica_etag_verification_enabled()
+    {
        return Ok(());
    }
    if single_part_replica_etag_mismatch(object_info.etag.as_deref(), response.etag.as_deref()) {
@@ -5870,6 +5877,57 @@ mod tests {
    use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
    use super::super::replication_object_decision_boundary::ReplicationMultipartPlanError;

+    #[test]
+    fn verify_single_part_replica_exempts_managed_sse_sources() {
+        // A new encrypted write stores a keyed fingerprint; a target that
+        // computes its own ETag answers with the MD5 of the plaintext it
+        // received. Both are 32 hex characters, so without the exemption the
+        // audit would fail every SSE-S3 / SSE-KMS replica (rustfs/backlog).
+        let fingerprint = "00112233445566778899aabbccddeeff";
+        let replica_md5 = "ffeeddccbbaa99887766554433221100";
+        let response = RemotePutObjectResponse {
+            version_id: None,
+            etag: Some(replica_md5.to_string()),
+        };
+        for sse in ["AES256", "aws:kms"] {
+            let object_info = ObjectInfo {
+                bucket: "source".to_string(),
+                name: "encrypted".to_string(),
+                etag: Some(fingerprint.to_string()),
+                user_defined: HashMap::from([("x-amz-server-side-encryption".to_string(), sse.to_string())]).into(),
+                ..Default::default()
+            };
+            verify_single_part_replica(&object_info, &response, false)
+                .unwrap_or_else(|err| panic!("{sse}: a managed-SSE source must not be audited against the replica MD5: {err}"));
+        }
+    }
+
+    #[test]
+    fn verify_single_part_replica_still_fails_plaintext_mismatch() {
+        if !replica_etag_verification_enabled() {
+            return;
+        }
+        let object_info = ObjectInfo {
+            bucket: "source".to_string(),
+            name: "plain".to_string(),
+            etag: Some("00112233445566778899aabbccddeeff".to_string()),
+            ..Default::default()
+        };
+        let mismatch = RemotePutObjectResponse {
+            version_id: None,
+            etag: Some("ffeeddccbbaa99887766554433221100".to_string()),
+        };
+        let err = verify_single_part_replica(&object_info, &mismatch, false)
+            .expect_err("a plaintext source whose replica MD5 differs must fail the replication");
+        assert_eq!(err.to_string(), REPLICA_ETAG_MISMATCH_ERROR);
+
+        let matching = RemotePutObjectResponse {
+            version_id: None,
+            etag: object_info.etag.clone(),
+        };
+        verify_single_part_replica(&object_info, &matching, false).expect("matching plaintext MD5s pass the audit");
+    }
+
    #[test]
    fn multipart_read_plan_preserves_legacy_plain_part_ranges() {
        const MIB: usize = 1024 * 1024;
@@ -157,6 +157,19 @@ pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap<String
    rustfs_replication::is_ssec_encrypted(user_defined)
 }

+/// Whether the stored ETag of a replication source can be compared with the
+/// ETag a target returns for the bytes it received. Only an unencrypted source
+/// stores the MD5 of the bytes that go on the wire. A managed-SSE (SSE-S3 /
+/// SSE-KMS) or SSE-C source stores a keyed fingerprint (new writes) or a legacy
+/// plaintext MD5, and the target re-encrypts what it receives, so neither
+/// equality nor inequality of the two ETags proves anything about the replica.
+pub(crate) fn replication_source_etag_is_comparable(user_defined: &HashMap<String, String>) -> bool {
+    matches!(
+        classify_replication_source_encryption(user_defined),
+        ReplicationSourceEncryption::Plaintext
+    )
+}
+
 /// HeadObjectOutput adapter over the pure SSE-C passthrough evidence
 /// judgment owned by `rustfs-replication`: extract the echoed
 /// customer-algorithm header and let the crate-owned policy decide.
@@ -578,6 +591,22 @@ mod tests {
    use time::Duration;
    use uuid::Uuid;

+    #[test]
+    fn replication_source_etag_is_comparable_only_for_plaintext_sources() {
+        assert!(replication_source_etag_is_comparable(&HashMap::new()));
+        for (key, value) in [
+            (metadata_keys::SERVER_SIDE_ENCRYPTION, "AES256"),
+            (metadata_keys::SERVER_SIDE_ENCRYPTION, "aws:kms"),
+            (SSEC_ALGORITHM_HEADER, "AES256"),
+        ] {
+            let metadata = HashMap::from([(key.to_string(), value.to_string())]);
+            assert!(
+                !replication_source_etag_is_comparable(&metadata),
+                "{key}={value} must not expose its stored ETag to the replica audit"
+            );
+        }
+    }
+
    /// backlog#1735 A3b SSE key-case finding. Every RustFS writer persists the
    /// SSE intent as lowercase `x-amz-server-side-encryption`
    /// (`encryption_material_to_metadata`; header-derived keys come from a
@@ -3691,6 +3691,55 @@ fn activation_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error {
    }
 }

+fn rebalance_worker_read_lock_error(err: rustfs_lock::LockError, object: &str) -> Error {
+    match err {
+        rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable {
+            mode: "read",
+            bucket: RUSTFS_META_BUCKET.to_string(),
+            object: object.to_string(),
+            required,
+            achieved,
+        },
+        other => Error::Lock(other),
+    }
+}
+
+/// Read-only admission to an already durable rebalance run. Keeping this
+/// separate from the activation writer prevents admission from authorizing a
+/// pool or rebalance metadata write under shared namespace locks.
+pub(crate) struct PoolRebalanceWorkerFence {
+    pool_meta_guard: rustfs_lock::NamespaceLockGuard,
+    rebalance_meta_guard: rustfs_lock::NamespaceLockGuard,
+    fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
+}
+
+impl PoolRebalanceWorkerFence {
+    pub(crate) fn set_fleet_proof(
+        &mut self,
+        fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
+    ) {
+        self.fleet_proof = fleet_proof;
+    }
+
+    pub(crate) fn ensure_held(&self) -> Result<()> {
+        if self.pool_meta_guard.is_lock_lost()
+            || self.pool_meta_guard.is_released()
+            || self.rebalance_meta_guard.is_lock_lost()
+            || self.rebalance_meta_guard.is_released()
+        {
+            return Err(Error::other("worker admission lock lost before rebalance activation"));
+        }
+        if self
+            .fleet_proof
+            .as_ref()
+            .is_some_and(|proof| !crate::services::notification_sys::cross_pool_fence_fleet_proof_matches(proof))
+        {
+            return Err(Error::other(POOL_ACTIVATION_FLEET_PROOF_EXPIRED));
+        }
+        Ok(())
+    }
+}
+
 pub(crate) struct PoolRebalanceActivationFence {
    pool_meta_guard: Arc<rustfs_lock::NamespaceLockGuard>,
    rebalance_meta_guard: Arc<rustfs_lock::NamespaceLockGuard>,
@@ -3762,6 +3811,35 @@ tokio::task_local! {
    pub(crate) static REBALANCE_ACTIVATION_LOCK_ATTEMPT: Arc<tokio::sync::Notify>;
 }

+pub(crate) async fn acquire_pool_rebalance_worker_locks<S>(pool: Arc<S>) -> Result<PoolRebalanceWorkerFence>
+where
+    S: crate::storage_api_contracts::namespace::NamespaceLocking<
+            Error = Error,
+            NamespaceLock = rustfs_lock::NamespaceLockWrapper,
+        >,
+{
+    // Match activation and entry admission: rebalance.bin -> pool.bin. Joining
+    // the same durable run must coexist with entries retaining their run fence.
+    let rebalance_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
+    #[cfg(test)]
+    let _ = REBALANCE_ACTIVATION_LOCK_ATTEMPT.try_with(|attempted| attempted.notify_one());
+    let rebalance_meta_guard = rebalance_meta_lock
+        .get_read_lock(get_lock_acquire_timeout())
+        .await
+        .map_err(|err| rebalance_worker_read_lock_error(err, REBAL_META_NAME))?;
+    let pool_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
+    let pool_meta_guard = pool_meta_lock
+        .get_read_lock(get_lock_acquire_timeout())
+        .await
+        .map_err(|err| rebalance_worker_read_lock_error(err, POOL_META_NAME))?;
+
+    Ok(PoolRebalanceWorkerFence {
+        pool_meta_guard,
+        rebalance_meta_guard,
+        fleet_proof: None,
+    })
+}
+
 pub(crate) async fn acquire_pool_rebalance_activation_locks<S>(
    pool: Arc<S>,
    fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
@@ -12430,6 +12508,18 @@ impl ECStore {
        Ok(pool_meta)
    }

+    pub(crate) async fn load_runtime_pool_meta_under_worker_fence(
+        &self,
+        write_state: &mut PoolMetaWriteState,
+        worker_fence: &PoolRebalanceWorkerFence,
+        operation: &str,
+    ) -> Result<PoolMeta> {
+        worker_fence.ensure_held()?;
+        let pool_meta = self.load_runtime_pool_meta_observing(write_state, operation).await?;
+        worker_fence.ensure_held()?;
+        Ok(pool_meta)
+    }
+
    pub(crate) async fn load_runtime_pool_meta_under_activation_fence(
        &self,
        write_state: &mut PoolMetaWriteState,
@@ -1006,6 +1006,9 @@ pub struct ObjectOptions {
    /// Proves an upper layer holds the bucket lifecycle sentinel. A separate
    /// fence avoids recursively acquiring the read lock behind a queued writer.
    pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
+    /// Retains the actual lifecycle guard for an owned multipart commit. A
+    /// lock-loss fence alone does not keep a cancelled caller's lock alive.
+    pub bucket_lifecycle_lock_owner: Option<Arc<rustfs_lock::NamespaceLockGuard>>,
    pub replication_request: bool,
    /// True when the inbound request carried the
    /// `{x-rustfs-,x-minio-}source-proxy-request` header family with the
@@ -1165,6 +1168,7 @@ impl std::fmt::Debug for ObjectOptions {
            .field("delete_mutation_owner", &self.delete_mutation_owner.is_some())
            .field("namespace_lock_fence", &self.namespace_lock_fence.is_some())
            .field("bucket_lifecycle_lock_fence", &self.bucket_lifecycle_lock_fence.is_some())
+            .field("bucket_lifecycle_lock_owner", &self.bucket_lifecycle_lock_owner.is_some())
            .field("replication_request", &self.replication_request)
            .field("proxy_request", &self.proxy_request)
            .field("proxy_header_set", &self.proxy_header_set)
@@ -18,8 +18,8 @@ use super::{
    encode_rebalance_stop_propagation_record,
 };
 use crate::core::pools::{
-    PoolMeta, PoolRebalanceActivationFence, acquire_pool_activation_fleet_proof, acquire_pool_rebalance_activation_locks,
-    pool_meta_has_active_decommission,
+    PoolMeta, PoolRebalanceActivationFence, PoolRebalanceWorkerFence, acquire_pool_activation_fleet_proof,
+    acquire_pool_rebalance_activation_locks, acquire_pool_rebalance_worker_locks, pool_meta_has_active_decommission,
 };
 use crate::error::{Error, Result};
 use crate::object_api::ObjectOptions;
@@ -77,10 +77,33 @@ fn ensure_rebalance_activation_pool_meta_allowed(meta: &PoolMeta) -> Result<()>
 }

 pub(super) enum RebalanceWorkerActivationFence {
-    Ready(Box<PoolRebalanceActivationFence>),
+    Ready(Box<PoolRebalanceWorkerFence>),
    NotStartedTerminal,
 }

+async fn persisted_rebalance_requires_worker_activation<S>(pool: Arc<S>, expected_id: &str) -> Result<bool>
+where
+    S: EcstoreObjectIO,
+{
+    let mut persisted = RebalanceMeta::new();
+    persisted
+        .load_with_opts(
+            pool,
+            ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+        )
+        .await?;
+    if persisted.id != expected_id {
+        return Err(Error::other(format!(
+            "rebalance metadata changed before worker activation: expected {expected_id}, found {}",
+            persisted.id
+        )));
+    }
+    Ok(crate::services::rebalance::rebalance_requires_worker_activation(&persisted))
+}
+
 pub(super) struct RebalanceRunGuard {
    _local_guard: tokio::sync::OwnedRwLockReadGuard<()>,
    persisted_guard: rustfs_lock::NamespaceLockGuard,
@@ -577,48 +600,58 @@ impl ECStore {
    where
        S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
    {
-        // Lock order: pool_meta_save_gate -> rebalance.bin -> pool.bin.
+        // Lock order: pool_meta_save_gate -> rebalance.bin -> pool.bin. These
+        // shared fences admit a peer without waiting for the run's own entries.
        let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
        pool_meta_guard.ensure_write_safe("rebalance worker activation")?;
-        // Classify the durable rebalance record while holding both namespace
-        // fences. A terminal record is a no-op and must not depend on the
-        // notification subsystem having published a fleet proof yet.
-        let mut activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), None).await?;
+        let mut worker_fence = acquire_pool_rebalance_worker_locks(pool.clone()).await?;
        let pool_meta = self
-            .load_runtime_pool_meta_under_activation_fence(&mut pool_meta_guard, &activation_fence, "rebalance worker activation")
+            .load_runtime_pool_meta_under_worker_fence(&mut pool_meta_guard, &worker_fence, "rebalance worker activation")
            .await?;
        ensure_rebalance_activation_pool_meta_allowed(&pool_meta)?;
-
-        let mut persisted = RebalanceMeta::new();
-        persisted
-            .load_with_opts(
-                pool,
-                ObjectOptions {
-                    no_lock: true,
-                    ..Default::default()
-                },
-            )
-            .await?;
-        if persisted.id != expected_id {
-            return Err(Error::other(format!(
-                "rebalance metadata changed before worker activation: expected {expected_id}, found {}",
-                persisted.id
-            )));
-        }
-
-        activation_fence.ensure_held()?;
-        if !crate::services::rebalance::rebalance_requires_worker_activation(&persisted) {
+        let needs_activation = persisted_rebalance_requires_worker_activation(pool, expected_id).await?;
+        worker_fence.ensure_held()?;
+        if !needs_activation {
            return Ok(RebalanceWorkerActivationFence::NotStartedTerminal);
        }

-        // Active worker admission still requires the fail-closed fleet proof.
-        // Attach it immediately before the final fence validation so expiry or
-        // topology changes are checked again at every later commit boundary.
+        let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?;
+        worker_fence.set_fleet_proof(fleet_proof);
+        worker_fence.ensure_held()?;
+        Ok(RebalanceWorkerActivationFence::Ready(Box::new(worker_fence)))
+    }
+
+    pub(super) async fn fence_rebalance_worker_metadata_update<S>(
+        &self,
+        pool: Arc<S>,
+        expected_id: &str,
+    ) -> Result<Option<Box<PoolRebalanceActivationFence>>>
+    where
+        S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
+    {
+        // The admission read fences must already be dropped. Reacquire in the
+        // original order and revalidate every durable boundary across that gap.
+        let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
+        pool_meta_guard.ensure_write_safe("rebalance worker metadata update")?;
+        let mut activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), None).await?;
+        let pool_meta = self
+            .load_runtime_pool_meta_under_activation_fence(
+                &mut pool_meta_guard,
+                &activation_fence,
+                "rebalance worker metadata update",
+            )
+            .await?;
+        ensure_rebalance_activation_pool_meta_allowed(&pool_meta)?;
+        let needs_activation = persisted_rebalance_requires_worker_activation(pool, expected_id).await?;
+        activation_fence.ensure_held()?;
+        if !needs_activation {
+            return Ok(None);
+        }
+
        let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?;
        activation_fence.set_fleet_proof(fleet_proof);
        activation_fence.ensure_held()?;
-
-        Ok(RebalanceWorkerActivationFence::Ready(Box::new(activation_fence)))
+        Ok(Some(Box::new(activation_fence)))
    }

    #[tracing::instrument(skip_all)]
@@ -1816,6 +1849,107 @@ mod tests {
        assert_activation_locks_released(&store).await;
    }

+    #[tokio::test]
+    #[serial_test::serial]
+    async fn rebalance_init_missing_fleet_proof_preserves_metadata_and_rechecks_competing_run() {
+        let previous_id = "completed-before-capability-refresh";
+        let terminal = RebalanceMeta {
+            id: previous_id.to_string(),
+            stopped_at: Some(OffsetDateTime::now_utc()),
+            pool_stats: vec![RebalanceStats::default(); 2],
+            ..Default::default()
+        };
+        let (_temp_dirs, store, competing_store) = crate::services::rebalance::test_two_pool_stores(Some(terminal.clone())).await;
+        let disk_stats = vec![
+            DiskStat {
+                total_space: 100,
+                available_space: 0,
+            },
+            DiskStat {
+                total_space: 100,
+                available_space: 100,
+            },
+        ];
+        let initial_epoch = store.ctx.data_movement_operation_epoch();
+        let proof_guard = crate::services::notification_sys::without_cross_pool_fence_fleet_proof_for_test();
+        set_rebalance_disk_stats_override_for_test(store.id, disk_stats.clone());
+        let err = store
+            .init_rebalance_start(vec!["waiting-request".to_string()])
+            .await
+            .expect_err("initialization must fail closed while the live proof is missing");
+        assert!(
+            err.to_string()
+                .contains("pool activation requires a live fleet capability proof")
+        );
+        assert!(
+            crate::core::pools::is_pool_activation_fleet_proof_error(&err),
+            "the admin handler retries init only on this error shape: {err}"
+        );
+        assert_eq!(store.ctx.data_movement_operation_epoch(), initial_epoch);
+        {
+            let local = store.rebalance_meta.read().await;
+            let local = local.as_ref().expect("failed initialization must preserve the previous run");
+            assert_eq!(local.id, previous_id);
+            assert_eq!(local.stopped_at, terminal.stopped_at);
+            assert!(local.cancel.is_none(), "failed initialization must not start a worker");
+        }
+        let mut persisted = RebalanceMeta::new();
+        persisted
+            .load(store.pools[0].clone())
+            .await
+            .expect("failed initialization must leave durable metadata readable");
+        assert_eq!(persisted.id, previous_id);
+        assert_eq!(persisted.stopped_at, terminal.stopped_at);
+
+        drop(proof_guard);
+        set_rebalance_disk_stats_override_for_test(competing_store.id, disk_stats.clone());
+        let winning_id = competing_store
+            .init_rebalance_start(vec!["winning-request".to_string()])
+            .await
+            .expect("a ready request must be able to initialize one run after the failed attempt");
+        assert_ne!(winning_id, previous_id);
+
+        // The waiting request still holds its old local snapshot. Its retry must
+        // recheck durable state under the activation locks before publishing.
+        set_rebalance_disk_stats_override_for_test(store.id, disk_stats);
+        let remote_conflict = store
+            .init_rebalance_start(vec!["waiting-request".to_string()])
+            .await
+            .expect_err("the waiting request must not replace a durable competing run");
+        assert!(
+            matches!(
+                crate::data_movement::data_movement_stage_source(&remote_conflict),
+                Some(Error::RebalanceAlreadyRunning)
+            ),
+            "the save-stage error must retain the typed competing-run conflict: {remote_conflict:?}"
+        );
+        let local_conflict = competing_store
+            .init_rebalance_start(vec!["another-request".to_string()])
+            .await
+            .expect_err("a local retry must not replace its active run");
+        assert!(
+            matches!(local_conflict, Error::RebalanceAlreadyRunning),
+            "the local admission check must report the competing-run conflict: {local_conflict:?}"
+        );
+        persisted
+            .load(store.pools[0].clone())
+            .await
+            .expect("the competing run must remain durable after both rejected retries");
+        assert_eq!(persisted.id, winning_id);
+        assert_eq!(persisted.pool_stats[0].buckets, vec!["winning-request".to_string()]);
+        assert_eq!(
+            store
+                .rebalance_meta
+                .read()
+                .await
+                .as_ref()
+                .expect("old local run must survive")
+                .id,
+            previous_id
+        );
+        assert_eq!(store.ctx.data_movement_operation_epoch(), initial_epoch + 1);
+    }
+
    #[tokio::test]
    #[serial_test::serial]
    async fn rebalance_worker_skips_terminal_metadata_without_fleet_proof() {
@@ -1874,6 +2008,128 @@ mod tests {
        );
    }

+    async fn assert_worker_write_upgrade_revalidates_durable_state(change: &str) {
+        let rebalance_id = "rebalance-worker-write-upgrade";
+        let active = RebalanceMeta {
+            id: rebalance_id.to_string(),
+            percent_free_goal: 0.5,
+            pool_stats: vec![
+                RebalanceStats {
+                    participating: true,
+                    init_capacity: 100,
+                    info: RebalanceInfo {
+                        status: RebalStatus::Started,
+                        ..Default::default()
+                    },
+                    ..Default::default()
+                },
+                RebalanceStats::default(),
+            ],
+            ..Default::default()
+        };
+        let (_temp_dirs, store, peer) =
+            crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(Some(active.clone())).await;
+        let arrived = Arc::new(tokio::sync::Notify::new());
+        let release = Arc::new(tokio::sync::Notify::new());
+        let worker = Arc::clone(&store);
+        let mut tasks = tokio::task::JoinSet::new();
+        tasks.spawn(
+            super::super::runtime::REBALANCE_WORKER_WRITE_UPGRADE_BARRIER
+                .scope((Arc::clone(&arrived), Arc::clone(&release)), async move {
+                    worker.start_rebalance_for_id(rebalance_id).await
+                }),
+        );
+        tokio::time::timeout(std::time::Duration::from_secs(30), arrived.notified())
+            .await
+            .expect("empty-queue activation must release its read fences before requesting write authority");
+
+        let mut changed = active.clone();
+        match change {
+            "stopped" => {
+                changed.stopped_at = Some(OffsetDateTime::now_utc());
+                changed.pool_stats[0].info.status = RebalStatus::Stopped;
+            }
+            "replaced" => changed.id = "replacement-rebalance-run".to_string(),
+            "decommission" => {
+                let mut pool_meta = peer.pool_meta.read().await.clone();
+                pool_meta.pools[0].decommission = Some(crate::core::pools::PoolDecommissionInfo {
+                    start_time: Some(OffsetDateTime::now_utc()),
+                    ..Default::default()
+                });
+                pool_meta
+                    .save(peer.pools.clone())
+                    .await
+                    .expect("a competing pool metadata writer must acquire the released pool fence");
+            }
+            _ => panic!("unknown durable state transition"),
+        }
+        changed
+            .save_with_opts(
+                peer.pools[0].clone(),
+                ObjectOptions {
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a competing run writer must acquire the released rebalance fence");
+        release.notify_one();
+        let result = tokio::time::timeout(std::time::Duration::from_secs(30), tasks.join_next())
+            .await
+            .expect("activation must finish after the durable transition")
+            .expect("activation task must be tracked")
+            .expect("activation must not panic");
+        match change {
+            "stopped" => result.expect("a durably stopped run must become an activation no-op"),
+            "replaced" => assert!(
+                result
+                    .expect_err("a replaced run must reject stale activation")
+                    .to_string()
+                    .contains("expected"),
+            ),
+            "decommission" => assert!(matches!(result, Err(Error::DecommissionAlreadyRunning))),
+            _ => unreachable!(),
+        }
+        let mut persisted = RebalanceMeta::new();
+        persisted
+            .load(peer.pools[0].clone())
+            .await
+            .expect("the competing durable state must remain readable");
+        assert_eq!(persisted.id, changed.id, "activation must not overwrite the run identity");
+        assert_eq!(persisted.stopped_at, changed.stopped_at, "activation must not overwrite the stop marker");
+        assert_eq!(persisted.pool_stats[0].info.status, changed.pool_stats[0].info.status);
+        assert!(
+            store
+                .rebalance_meta
+                .read()
+                .await
+                .as_ref()
+                .expect("local metadata must remain present")
+                .cancel
+                .is_none(),
+            "rejected activation must not install a worker token",
+        );
+        assert_activation_locks_released(&store).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn rebalance_worker_write_upgrade_rechecks_stop() {
+        assert_worker_write_upgrade_revalidates_durable_state("stopped").await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn rebalance_worker_write_upgrade_rechecks_run_identity() {
+        assert_worker_write_upgrade_revalidates_durable_state("replaced").await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn rebalance_worker_write_upgrade_rechecks_decommission() {
+        assert_worker_write_upgrade_revalidates_durable_state("decommission").await;
+    }
+
    #[tokio::test]
    #[serial_test::serial]
    async fn rebalance_activation_adopts_commit_after_post_save_fence_loss() {
@@ -1440,7 +1440,7 @@ mod tests {

    #[tokio::test]
    #[serial_test::serial]
-    async fn real_rebalance_entry_progresses_while_peer_activation_waits_for_run_fence() {
+    async fn real_rebalance_peer_activates_while_entry_retains_run_fence() {
        const REBALANCE_ID: &str = "rebalance-peer-activation-lock-order";
        let (_temp_dirs, store, peer) = crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(Some(
            active_rebalance_meta(REBALANCE_ID),
@@ -1456,7 +1456,7 @@ mod tests {
        let bucket = crate::disk::RUSTFS_META_BUCKET;
        let object = "rebalance-peer-activation-object";
        let version_id = uuid::Uuid::new_v4();
-        let payload = b"entry must drain before peer activation takes the pool fence".repeat(1024);
+        let payload = b"peer activation must not wait for this entry to drain".repeat(1024);
        let source_set = store.pools[0].get_disks_by_key(object);
        let target_set = store.pools[1].get_disks_by_key(object);
        let opts = ObjectOptions {
@@ -1516,7 +1516,12 @@ mod tests {
        );
        tokio::time::timeout(StdDuration::from_secs(30), attempted.notified())
            .await
-            .expect("peer activation must attempt the persisted rebalance write fence");
+            .expect("peer activation must attempt the persisted rebalance fence");
+        tokio::time::timeout(StdDuration::from_secs(30), activation_result)
+            .await
+            .expect("peer activation must finish while the entry remains paused")
+            .expect("peer activation result should be sent")
+            .expect("joining the same run must not require its active entry to release the run fence");
        release.notify_one();

        tokio::time::timeout(StdDuration::from_secs(30), async {
@@ -1531,10 +1536,6 @@ mod tests {
        })
        .await
        .expect("entry and peer activation must both make progress");
-        activation_result
-            .await
-            .expect("peer activation result should be sent")
-            .expect("peer activation must not time out behind the entry it blocks");

        let mut reader = target_set
            .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
@@ -27,6 +27,11 @@ use tokio::time::{Duration, Instant};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};

+#[cfg(test)]
+tokio::task_local! {
+    pub(super) static REBALANCE_WORKER_WRITE_UPGRADE_BARRIER: (Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>);
+}
+
 pub(super) fn should_fail_repeated_rebalance_bucket_defer(
    deferred_buckets: &mut HashSet<String>,
    bucket: &str,
@@ -213,6 +218,22 @@ impl ECStore {
        }

        if must_persist {
+            // Shared admission grants no write authority. Release it before
+            // taking the exclusive metadata fences, then recheck stop/run/pool
+            // state before publishing the locally staged terminal candidate.
+            drop(activation_fence);
+            #[cfg(test)]
+            if let Ok((arrived, release)) = REBALANCE_WORKER_WRITE_UPGRADE_BARRIER.try_with(Clone::clone) {
+                arrived.notify_one();
+                release.notified().await;
+            }
+            let Some(activation_fence) = self
+                .fence_rebalance_worker_metadata_update(pool.clone(), expected_id.as_ref())
+                .await?
+            else {
+                cancel_tx.cancel();
+                return Ok(false);
+            };
            let save_result = resolve_rebalance_meta_save_result(
                self.save_rebalance_meta_under_activation_fence(
                    pool,
@@ -242,13 +263,14 @@ impl ECStore {
                cancel_tx.cancel();
                return Err(err);
            }
+        } else {
+            if let Err(err) = activation_fence.ensure_held() {
+                let mut rebalance_meta = self.rebalance_meta.write().await;
+                rollback_local_rebalance_worker_activation(rebalance_meta.as_mut(), expected_id.as_ref(), &rx);
+                return Err(err);
+            }
+            drop(activation_fence);
        }
-        if !must_persist && let Err(err) = activation_fence.ensure_held() {
-            let mut rebalance_meta = self.rebalance_meta.write().await;
-            rollback_local_rebalance_worker_activation(rebalance_meta.as_mut(), expected_id.as_ref(), &rx);
-            return Err(err);
-        }
-        drop(activation_fence);

        if activation_outcome != RebalanceLocalActivationOutcome::Started {
            return Ok(must_persist);
@@ -63,7 +63,6 @@ use crate::crash_inject::{self, CrashPoint};
 use crate::disk::DiskAPI;
 #[cfg(test)]
 use crate::disk::DiskOption;
-#[cfg(test)]
 use crate::disk::STORAGE_FORMAT_FILE;
 #[cfg(test)]
 use crate::disk::new_disk;
@@ -231,6 +230,60 @@ pub(crate) struct StaleMultipartCleanupGuard {
    lock_guard: ObjectLockDiagGuard,
 }

+/// Exact-key recovery is admitted only after every disk proves that the upload
+/// metadata is absent. Keep the checked disk generation and upload lock through
+/// deletion; a quorum miss is not proof that an upload is an orphan.
+/// Final object data, completion replay identity, and old-data cleanup receipts
+/// live under the destination bucket/object, never this staging path. This also
+/// permits removing an exact orphan from a previous bucket incarnation without
+/// claiming that its UploadID belongs to the current incarnation.
+pub(crate) struct OrphanMultipartCleanupGuard {
+    set: Arc<SetDisks>,
+    disks: Vec<Option<DiskStore>>,
+    upload_path: String,
+    lock_guard: Arc<ObjectLockDiagGuard>,
+    exists: bool,
+}
+
+impl OrphanMultipartCleanupGuard {
+    pub(crate) async fn delete(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<bool> {
+        fence_commit_on_lock_loss(Some(&self.lock_guard), "orphan_multipart_cleanup", &self.upload_path)?;
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
+        if !self.exists {
+            return Ok(false);
+        }
+
+        let results = join_all(self.disks.iter().map(|disk| async {
+            let Some(disk) = disk else {
+                return Err(DiskError::DiskNotFound);
+            };
+            match disk
+                .delete(
+                    RUSTFS_META_MULTIPART_BUCKET,
+                    &self.upload_path,
+                    crate::disk::DeleteOptions {
+                        recursive: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+            {
+                Err(DiskError::FileNotFound | DiskError::PathNotFound | DiskError::VolumeNotFound) => Ok(()),
+                result => result,
+            }
+        }))
+        .await;
+        self.set.record_capacity_scope_if_needed(None, &self.disks);
+        let errors = results.into_iter().map(|result| result.err()).collect::<Vec<_>>();
+        if let Some(err) = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, self.disks.len()) {
+            return Err(to_object_err(err.into(), vec![bucket, object]));
+        }
+        fence_commit_on_lock_loss(Some(&self.lock_guard), "orphan_multipart_cleanup", &self.upload_path)?;
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
+        Ok(true)
+    }
+}
+
 pub(crate) struct DataMovementMultipartAbortGuard {
    upload_path: String,
    write_quorum: Option<usize>,
@@ -557,6 +610,34 @@ fn multipart_bucket_incarnation_matches(metadata: &HashMap<String, String>, expe
    matches!(multipart_bucket_incarnation_id(metadata), Ok(Some(actual)) if actual == expected)
 }

+fn orphan_multipart_upload_id_is_canonical(upload_id: &str) -> bool {
+    let Ok(decoded) = base64_simd::URL_SAFE_NO_PAD.decode_to_vec(upload_id.as_bytes()) else {
+        return false;
+    };
+    let Ok(decoded) = std::str::from_utf8(&decoded) else {
+        return false;
+    };
+    let Some((_, raw)) = decoded.split_once('.') else {
+        return false;
+    };
+    let Some((uuid, timestamp)) = raw.split_once('x') else {
+        return false;
+    };
+    let Ok(id) = Uuid::parse_str(uuid) else {
+        return false;
+    };
+    if id.is_nil() || id.to_string() != uuid {
+        return false;
+    }
+    let Ok(timestamp_ns) = timestamp.parse::<i128>() else {
+        return false;
+    };
+    if timestamp_ns.to_string() != timestamp {
+        return false;
+    }
+    OffsetDateTime::from_unix_timestamp_nanos(timestamp_ns).is_ok()
+}
+
 fn validate_multipart_bucket_incarnation(
    metadata: &HashMap<String, String>,
    bucket: &str,
@@ -1246,6 +1327,114 @@ impl SetDisks {
        })
    }

+    pub(crate) async fn lock_orphan_multipart_cleanup(
+        self: &Arc<Self>,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        opts: &ObjectOptions,
+        held_guards: &[OrphanMultipartCleanupGuard],
+    ) -> Result<OrphanMultipartCleanupGuard> {
+        let invalid_upload = || StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned());
+        if opts.data_movement
+            || opts.no_lock
+            || self.local_lock_manager.is_disabled()
+            || !orphan_multipart_upload_id_is_canonical(upload_id)
+        {
+            return Err(invalid_upload());
+        }
+        // This path is derived from the caller's exact bucket/key and a canonical
+        // UUID/timestamp upload identity, never from the shared MPU directory list.
+        let upload_path = Self::get_upload_id_dir(bucket, object, upload_id);
+        // Pool labels do not partition the raw ObjectKey sent to lock clients.
+        // Reuse the actual owner only for this exact path and proven backend;
+        // independent distributed domains still need their own guards.
+        let distributed = self.ctx.is_dist_erasure().await;
+        let mut lock_guard = None;
+        for held in held_guards {
+            if held.upload_path != upload_path {
+                continue;
+            }
+            let shared_domain = match (distributed, held.set.ctx.is_dist_erasure().await) {
+                (false, false) => self.shares_namespace_lock_domain(&held.set).await,
+                (true, true) => {
+                    let same = super::super::same_distributed_lock_domain(&self.lockers, &held.set.lockers);
+                    if !same
+                        && self
+                            .lockers
+                            .iter()
+                            .any(|client| held.set.lockers.iter().any(|held_client| Arc::ptr_eq(client, held_client)))
+                    {
+                        // A guard for only part of another quorum is not proof
+                        // of ownership there, and acquiring it again may wait
+                        // on ourselves. Preserve all staging in this topology.
+                        return Err(Error::other("orphan multipart cleanup cannot cover partially overlapping lock domains"));
+                    }
+                    same
+                }
+                _ => false,
+            };
+            if shared_domain {
+                lock_guard = Some(Arc::clone(&held.lock_guard));
+                break;
+            }
+        }
+        let lock_guard = match lock_guard {
+            Some(guard) => guard,
+            None => Arc::new(
+                self.acquire_write_lock_diag("orphan_multipart_cleanup", RUSTFS_META_MULTIPART_BUCKET, &upload_path)
+                    .await?,
+            ),
+        };
+        fence_commit_on_lock_loss(Some(&lock_guard), "orphan_multipart_cleanup", &upload_path)?;
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
+        let disks = self.disks.read().await.clone();
+        if disks.len() != self.set_drive_count || disks.is_empty() {
+            return Err(Error::ErasureReadQuorum);
+        }
+        // Raw full fan-out cannot quorum-short-circuit or interpret an empty
+        // version list as absence of the physical metadata file.
+        let (_, errors) = Self::read_all_raw_file_info(&disks, RUSTFS_META_MULTIPART_BUCKET, &upload_path, false).await;
+        if errors.len() != disks.len() {
+            return Err(Error::ErasureReadQuorum);
+        }
+        for error in errors {
+            match error {
+                Some(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
+                // Even one valid metadata copy may still be recoverable. Never
+                // convert a partial, corrupt, or unreachable upload into debris.
+                None => return Err(invalid_upload()),
+                Some(error) => return Err(to_object_err(error.into(), vec![bucket, object, upload_id])),
+            }
+        }
+        let mut exists = false;
+        for disk in &disks {
+            let Some(disk) = disk else {
+                return Err(Error::ErasureReadQuorum);
+            };
+            match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await {
+                Ok(entries) => {
+                    // Raw metadata reads report a directory named xl.meta as
+                    // FileNotFound. Its physical presence is unknown/corrupt
+                    // metadata, not proof that this staging can be deleted.
+                    if entries.iter().any(|entry| entry.trim_end_matches('/') == STORAGE_FORMAT_FILE) {
+                        return Err(to_object_err(DiskError::FileCorrupt.into(), vec![bucket, object, upload_id]));
+                    }
+                    exists = true;
+                }
+                Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
+                Err(error) => return Err(to_object_err(error.into(), vec![bucket, object, upload_id])),
+            }
+        }
+        Ok(OrphanMultipartCleanupGuard {
+            set: self.clone(),
+            disks,
+            upload_path,
+            lock_guard,
+            exists,
+        })
+    }
+
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn list_multipart_uploads_for_incarnation(
        &self,
@@ -1678,6 +1867,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
            drop(admission_guard);
        }

+        let mut commit_owns_cleanup = false;
        let result: Result<PartInfo> = async {
            let erasure =
                Arc::new(coding::Erasure::try_new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size)
@@ -1881,120 +2071,171 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                    .await?;
                (Some(upload_guard), Some(part_guard))
            };
-            let (commit_fi, _) = self
-                .check_upload_id_exists_with_opts(bucket, object, upload_id, multipart_size_limit.is_some(), opts)
-                .await?;
-            let commit_size_limit = multipart_size_limit_from_metadata(&commit_fi.metadata)?;
-            if commit_size_limit != multipart_size_limit {
-                return Err(Error::InvalidArgument(
-                    "multipart upload".to_string(),
-                    SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
-                    "size limit metadata changed or is missing".to_string(),
-                ));
-            }
-            ensure_data_movement_upload_access(&commit_fi, bucket, object, upload_id, opts)?;
-            ensure_multipart_bucket_incarnation(
-                &self.ctx,
-                &commit_fi,
-                bucket,
-                object,
-                upload_id,
-                opts.expected_bucket_incarnation_id,
-            )
-            .await?;
-            #[cfg(test)]
-            pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await;
-            fence_commit_on_lock_loss(_upload_commit_guard.as_ref(), "put_object_part_commit", &upload_id_path)?;
-            fence_commit_on_lock_loss(_part_commit_guard.as_ref(), "put_object_part_commit", &part_lock_path)?;
-            if opts
-                .namespace_lock_fence
-                .as_ref()
-                .is_some_and(NamespaceLockFence::is_lock_lost)
-            {
-                return Err(StorageError::NamespaceLockQuorumUnavailable {
-                    mode: "put_object_part_outer_lock",
-                    bucket: bucket.to_string(),
-                    object: object.to_string(),
-                    required: 1,
-                    achieved: 0,
-                });
-            }
-            ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
-
-            if let Some(limit) = commit_size_limit {
-                let current_size = self
-                    .current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &commit_fi, part_id)
-                    .await?;
-                let candidate_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
-                admitted_multipart_size(current_size, candidate_size, limit)?;
-            }
-
-            let decommission_capacity_guard = if let Some(store) = opts.decommission_capacity_admission.as_ref() {
-                Some(
-                    store
-                        .acquire_external_decommission_capacity_fence(
-                            &[self.pool_index],
-                            DecommissionCapacityAdmission::ExistingMultipart,
-                        )
-                        .await?,
-                )
-            } else {
-                None
-            };
-            #[cfg(test)]
-            pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterCapacityAdmission).await;
-            if decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
-                return Err(StorageError::NamespaceLockQuorumUnavailable {
-                    mode: "put_object_part_decommission_capacity",
-                    bucket: bucket.to_string(),
-                    object: object.to_string(),
-                    required: 1,
-                    achieved: 0,
-                });
-            }
-            let rename_result = self
-                .rename_part(
-                    &shuffle_disks,
-                    RUSTFS_META_TMP_BUCKET,
-                    &tmp_part_path,
-                    RUSTFS_META_MULTIPART_BUCKET,
-                    &part_path,
-                    part_info_buff.into(),
-                    write_quorum,
-                    Some(MultipartWriteQuorumContext {
-                        stage: MULTIPART_WRITE_QUORUM_RENAME_PART,
+            // Only guards owned by this operation may outlive its caller. Internal
+            // data-movement and borrowed-lock callers retain their synchronous
+            // lifetime; a fence by itself cannot transfer the outer lock lease.
+            let detach_commit_owner = _upload_commit_guard.is_some()
+                && opts.namespace_lock_fence.is_none()
+                && (opts.bucket_lifecycle_lock_fence.is_none() || opts.bucket_lifecycle_lock_owner.is_some());
+            let commit_set = self.clone();
+            let commit_bucket = bucket.to_owned();
+            let commit_object = object.to_owned();
+            let commit_upload_id = upload_id.to_owned();
+            let commit_upload_id_path = upload_id_path.clone();
+            let commit_tmp_part = tmp_part.clone();
+            let commit_opts = opts.clone();
+            commit_owns_cleanup = true;
+            // The body is already staged. Keep every commit guard through the
+            // prepare, publish, settlement/rollback and temporary cleanup I/O,
+            // even when the HTTP request stops awaiting this task.
+            let commit = async move {
+                let _upload_commit_guard = _upload_commit_guard;
+                let _part_commit_guard = _part_commit_guard;
+                let _capped_staging_guard = _capped_staging_guard;
+                let bucket = commit_bucket.as_str();
+                let object = commit_object.as_str();
+                let upload_id = commit_upload_id.as_str();
+                let upload_id_path = commit_upload_id_path;
+                let opts = &commit_opts;
+                let result: Result<PartInfo> = async {
+                    let (commit_fi, _) = commit_set
+                        .check_upload_id_exists_with_opts(bucket, object, upload_id, multipart_size_limit.is_some(), opts)
+                        .await?;
+                    let commit_size_limit = multipart_size_limit_from_metadata(&commit_fi.metadata)?;
+                    if commit_size_limit != multipart_size_limit {
+                        return Err(Error::InvalidArgument(
+                            "multipart upload".to_string(),
+                            SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
+                            "size limit metadata changed or is missing".to_string(),
+                        ));
+                    }
+                    ensure_data_movement_upload_access(&commit_fi, bucket, object, upload_id, opts)?;
+                    ensure_multipart_bucket_incarnation(
+                        &commit_set.ctx,
+                        &commit_fi,
                        bucket,
                        object,
                        upload_id,
-                        part_number: Some(part_id),
-                    }),
-                )
+                        opts.expected_bucket_incarnation_id,
+                    )
+                    .await?;
+                    #[cfg(test)]
+                    pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await;
+                    fence_commit_on_lock_loss(_upload_commit_guard.as_ref(), "put_object_part_commit", &upload_id_path)?;
+                    fence_commit_on_lock_loss(_part_commit_guard.as_ref(), "put_object_part_commit", &part_lock_path)?;
+                    if opts
+                        .namespace_lock_fence
+                        .as_ref()
+                        .is_some_and(NamespaceLockFence::is_lock_lost)
+                    {
+                        return Err(StorageError::NamespaceLockQuorumUnavailable {
+                            mode: "put_object_part_outer_lock",
+                            bucket: bucket.to_string(),
+                            object: object.to_string(),
+                            required: 1,
+                            achieved: 0,
+                        });
+                    }
+                    ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
+
+                    if let Some(limit) = commit_size_limit {
+                        let current_size = commit_set
+                            .current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &commit_fi, part_id)
+                            .await?;
+                        let candidate_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
+                        admitted_multipart_size(current_size, candidate_size, limit)?;
+                    }
+
+                    let decommission_capacity_guard = if let Some(store) = opts.decommission_capacity_admission.as_ref() {
+                        Some(
+                            store
+                                .acquire_external_decommission_capacity_fence(
+                                    &[commit_set.pool_index],
+                                    DecommissionCapacityAdmission::ExistingMultipart,
+                                )
+                                .await?,
+                        )
+                    } else {
+                        None
+                    };
+                    #[cfg(test)]
+                    pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterCapacityAdmission).await;
+                    if decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+                        return Err(StorageError::NamespaceLockQuorumUnavailable {
+                            mode: "put_object_part_decommission_capacity",
+                            bucket: bucket.to_string(),
+                            object: object.to_string(),
+                            required: 1,
+                            achieved: 0,
+                        });
+                    }
+                    let rename_result = commit_set
+                        .rename_part(
+                            &shuffle_disks,
+                            RUSTFS_META_TMP_BUCKET,
+                            &tmp_part_path,
+                            RUSTFS_META_MULTIPART_BUCKET,
+                            &part_path,
+                            part_info_buff.into(),
+                            write_quorum,
+                            Some(MultipartWriteQuorumContext {
+                                stage: MULTIPART_WRITE_QUORUM_RENAME_PART,
+                                bucket,
+                                object,
+                                upload_id,
+                                part_number: Some(part_id),
+                            }),
+                        )
+                        .await;
+                    drop(decommission_capacity_guard);
+                    let _ = rename_result?;
+                    #[cfg(test)]
+                    observe_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
+
+                    #[cfg(test)]
+                    pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterRename).await;
+
+                    let ret: PartInfo = PartInfo {
+                        etag: Some(etag.clone()),
+                        part_num: part_id,
+                        last_mod: Some(OffsetDateTime::now_utc()),
+                        size: w_size,
+                        actual_size,
+                    };
+
+                    // error!("put_object_part ret {:?}", &ret);
+
+                    Ok(ret)
+                }
                .await;
-            drop(decommission_capacity_guard);
-            let _ = rename_result?;
-            #[cfg(test)]
-            observe_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
-
-            #[cfg(test)]
-            pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterRename).await;
-            drop(_part_commit_guard);
-            drop(_upload_commit_guard);
-
-            let ret: PartInfo = PartInfo {
-                etag: Some(etag.clone()),
-                part_num: part_id,
-                last_mod: Some(OffsetDateTime::now_utc()),
-                size: w_size,
-                actual_size,
+                if result.is_err()
+                    && let Err(err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_part).await
+                {
+                    warn!(
+                        event = EVENT_SET_DISK_MULTIPART,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        result = "cleanup_failed",
+                        bucket,
+                        object,
+                        error = ?err,
+                        "failed to clean multipart commit staging"
+                    );
+                }
+                result
            };
-
-            // error!("put_object_part ret {:?}", &ret);
-
-            Ok(ret)
+            if detach_commit_owner {
+                tokio::spawn(commit)
+                    .await
+                .map_err(|err| Error::other_with_context("put_object_part commit task failed", err))?
+            } else {
+                commit.await
+            }
        }
        .await;

-        if result.is_err()
+        if !commit_owns_cleanup
+            && result.is_err()
            && let Err(err) = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_part).await
        {
            warn!(tmp_part = %tmp_part, error = ?err, "failed to cleanup multipart temporary data");
@@ -4233,6 +4474,300 @@ mod tests {
        }
    }

+    #[test]
+    fn orphan_multipart_identity_requires_canonical_upload_id() {
+        let created = OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(10);
+        let id = Uuid::new_v4();
+        let timestamp = (created + time::Duration::seconds(1)).unix_timestamp_nanos();
+        let encode = |raw: &str| runtime_sources::deployment_upload_id(raw);
+        let upload_id = encode(&format!("{id}x{timestamp}"));
+        assert!(orphan_multipart_upload_id_is_canonical(&upload_id));
+        for raw in [
+            format!("{}x{timestamp}", Uuid::nil()),
+            format!("{}x{timestamp}", id.simple()),
+            format!("{id}x0{timestamp}"),
+            format!("{id}x+{timestamp}"),
+            format!("{id}x{timestamp}/../another-upload"),
+            format!("../{id}x{timestamp}"),
+            format!("{id}x{}", i128::MAX),
+        ] {
+            assert!(
+                !orphan_multipart_upload_id_is_canonical(&encode(&raw)),
+                "noncanonical upload identity must be rejected: {raw}"
+            );
+        }
+        assert!(!orphan_multipart_upload_id_is_canonical("not-base64"));
+    }
+
+    #[tokio::test]
+    async fn orphan_multipart_cleanup_requires_every_disk_to_prove_metadata_absence() {
+        let (_temp_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "orphan-multipart-proof";
+        let object = "object";
+        make_bucket_on_all(&disks, bucket).await;
+        let upload = set
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("create upload");
+        let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id);
+        let metadata_path = format!("{upload_path}/{STORAGE_FORMAT_FILE}");
+        let leftover_path = format!("{upload_path}/part.late");
+        for disk in &disks {
+            disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path, Bytes::from_static(b"late part"))
+                .await
+                .expect("write late part fixture");
+        }
+        for disk in &disks[..3] {
+            disk.delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
+                .await
+                .expect("remove majority metadata");
+        }
+        let opts = ObjectOptions::default();
+        assert!(
+            set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
+                .await
+                .is_err(),
+            "one surviving metadata copy must block orphan recovery"
+        );
+        disks[3]
+            .write_all(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Bytes::from_static(b"corrupt metadata"))
+            .await
+            .expect("corrupt final metadata copy");
+        assert!(
+            set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
+                .await
+                .is_err(),
+            "corrupt metadata must not be treated as absence"
+        );
+        disks[3]
+            .delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
+            .await
+            .expect("remove final metadata copy");
+        set.disks.write().await[3] = None;
+        assert!(
+            set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
+                .await
+                .is_err(),
+            "an unreachable disk must not be hidden by absence quorum"
+        );
+        set.disks.write().await[3] = Some(disks[3].clone());
+        for disk in &disks {
+            assert_eq!(
+                disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path)
+                    .await
+                    .expect("failed preflight must preserve every part"),
+                Bytes::from_static(b"late part")
+            );
+        }
+
+        let guard = set
+            .lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
+            .await
+            .expect("all disks now prove metadata absence");
+        assert!(guard.delete(bucket, object, &opts).await.expect("recover exact staging"));
+        drop(guard);
+        for disk in &disks {
+            assert!(matches!(
+                disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await,
+                Err(DiskError::FileNotFound | DiskError::VolumeNotFound)
+            ));
+        }
+        let guard = set
+            .lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
+            .await
+            .expect("already absent staging remains provably absent");
+        assert!(
+            !guard
+                .delete(bucket, object, &opts)
+                .await
+                .expect("absence does not report recovery")
+        );
+    }
+
+    #[tokio::test]
+    async fn orphan_multipart_cleanup_rejects_metadata_directory_on_one_disk() {
+        let (temp_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "orphan-multipart-directory";
+        let object = "object";
+        make_bucket_on_all(&disks, bucket).await;
+        let upload = set
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("create upload");
+        let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id);
+        let metadata_path = format!("{upload_path}/{STORAGE_FORMAT_FILE}");
+        let leftover_path = format!("{upload_path}/part.late");
+        for disk in &disks {
+            disk.delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
+                .await
+                .expect("remove upload metadata");
+            disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path, Bytes::from_static(b"late part"))
+                .await
+                .expect("write leftover part");
+        }
+        let metadata_directory = temp_dirs[3].path().join(RUSTFS_META_MULTIPART_BUCKET).join(&metadata_path);
+        tokio::fs::create_dir_all(&metadata_directory)
+            .await
+            .expect("replace one missing metadata file with a directory");
+        let (_, errors) =
+            SetDisks::read_all_raw_file_info(&set.disks.read().await, RUSTFS_META_MULTIPART_BUCKET, &upload_path, false).await;
+        assert!(
+            errors.iter().all(|error| matches!(error, Some(DiskError::FileNotFound))),
+            "raw metadata reads alone must reproduce the directory-as-absence case: {errors:?}"
+        );
+        assert!(
+            set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &ObjectOptions::default(), &[])
+                .await
+                .is_err(),
+            "one physical metadata directory must block cleanup on every disk"
+        );
+        assert!(metadata_directory.is_dir(), "failed preflight must preserve the metadata directory");
+        for disk in &disks {
+            assert_eq!(
+                disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path)
+                    .await
+                    .expect("failed preflight must preserve every leftover part"),
+                Bytes::from_static(b"late part")
+            );
+        }
+    }
+
+    #[derive(Clone, Copy, PartialEq, Eq)]
+    enum OrphanMultipartLockDomainCase {
+        Shared,
+        Independent,
+        PartiallyOverlapping,
+    }
+
+    async fn check_orphan_multipart_distributed_lock_owners(case: OrphanMultipartLockDomainCase) {
+        let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        ctx.update_erasure_type(SetupType::DistErasure).await;
+        let first_client: Arc<dyn LockClient> =
+            Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        let second_client: Arc<dyn LockClient> =
+            Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
+        let second_clients = match case {
+            OrphanMultipartLockDomainCase::Shared => vec![first_client.clone()],
+            OrphanMultipartLockDomainCase::Independent => vec![second_client],
+            OrphanMultipartLockDomainCase::PartiallyOverlapping => vec![first_client.clone(), second_client],
+        };
+        let (first_dirs, first_disks, first_set) =
+            hermetic_set_disks_with_lockers_and_ctx(4, 0, 2, vec![first_client], ctx.clone()).await;
+        let (second_dirs, second_disks, second_set) = hermetic_set_disks_with_lockers_and_ctx(4, 1, 2, second_clients, ctx).await;
+        let bucket = "orphan-multipart-dist-locks";
+        let object = "object";
+        make_bucket_on_all(&first_disks, bucket).await;
+        make_bucket_on_all(&second_disks, bucket).await;
+        let upload_id = runtime_sources::deployment_upload_id(&format!(
+            "{}x{}",
+            Uuid::new_v4(),
+            OffsetDateTime::now_utc().unix_timestamp_nanos()
+        ));
+        let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+        for directory in first_dirs.iter().chain(second_dirs.iter()) {
+            let staging = directory.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_path);
+            tokio::fs::create_dir_all(&staging)
+                .await
+                .expect("create exact orphan staging");
+            tokio::fs::write(staging.join("part.late"), b"late part")
+                .await
+                .expect("write orphan part");
+        }
+        let opts = ObjectOptions::default();
+        let first_guard = first_set
+            .lock_orphan_multipart_cleanup(bucket, object, &upload_id, &opts, &[])
+            .await
+            .expect("first domain has an orphan");
+        let second_result = tokio::time::timeout(
+            Duration::from_secs(10),
+            second_set.lock_orphan_multipart_cleanup(bucket, object, &upload_id, &opts, std::slice::from_ref(&first_guard)),
+        )
+        .await
+        .expect("lock-domain admission must not wait on a guard it already owns");
+        if case == OrphanMultipartLockDomainCase::PartiallyOverlapping {
+            let error = match second_result {
+                Ok(_) => panic!("one existing quorum cannot authorize a partially overlapping domain"),
+                Err(error) => error,
+            };
+            assert!(error.to_string().contains("partially overlapping lock domains"));
+            for directory in first_dirs.iter().chain(second_dirs.iter()) {
+                assert!(
+                    directory
+                        .path()
+                        .join(RUSTFS_META_MULTIPART_BUCKET)
+                        .join(&upload_path)
+                        .join("part.late")
+                        .exists(),
+                    "uncertain lock coverage must preserve every physical part"
+                );
+            }
+            return;
+        }
+        let second_guard = second_result.expect("the second pool has a proven lock domain");
+        assert_eq!(
+            Arc::ptr_eq(&first_guard.lock_guard, &second_guard.lock_guard),
+            case == OrphanMultipartLockDomainCase::Shared,
+            "only the same exact backend may share a real upload guard across pool labels"
+        );
+        let second_namespace = second_set
+            .new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_path)
+            .await
+            .unwrap();
+        let mut second_writer = Box::pin(second_namespace.get_write_lock(Duration::from_secs(10)));
+        assert!(futures::poll!(&mut second_writer).is_pending());
+        if case == OrphanMultipartLockDomainCase::Shared {
+            drop(first_guard);
+            assert!(
+                futures::poll!(&mut second_writer).is_pending(),
+                "the second pool must retain the shared real owner after the first guard drops"
+            );
+            drop(second_guard);
+            drop(
+                tokio::time::timeout(Duration::from_secs(10), second_writer)
+                    .await
+                    .expect("the shared backend should release after the last owner")
+                    .expect("shared-domain writer resumes"),
+            );
+            return;
+        }
+        let first_namespace = first_set
+            .new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_path)
+            .await
+            .unwrap();
+        let mut first_writer = Box::pin(first_namespace.get_write_lock(Duration::from_secs(10)));
+        assert!(futures::poll!(&mut first_writer).is_pending());
+        drop(first_guard);
+        drop(
+            tokio::time::timeout(Duration::from_secs(10), first_writer)
+                .await
+                .expect("first domain should release independently")
+                .expect("first writer resumes"),
+        );
+        assert!(futures::poll!(&mut second_writer).is_pending(), "the second real owner must remain held");
+        drop(second_guard);
+        drop(
+            tokio::time::timeout(Duration::from_secs(10), second_writer)
+                .await
+                .expect("second domain should release with its owner")
+                .expect("second writer resumes"),
+        );
+    }
+
+    #[tokio::test]
+    async fn orphan_multipart_cleanup_keeps_separate_distributed_lock_owners() {
+        check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::Independent).await;
+    }
+
+    #[tokio::test]
+    async fn orphan_multipart_cleanup_reuses_shared_distributed_owner_across_pools() {
+        check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::Shared).await;
+    }
+
+    #[tokio::test]
+    async fn orphan_multipart_cleanup_rejects_partially_overlapping_distributed_domains() {
+        check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::PartiallyOverlapping).await;
+    }
+
    async fn stage_upload_with_create_opts(
        set_disks: &Arc<SetDisks>,
        bucket: &str,
@@ -6370,6 +6905,76 @@ mod tests {
            .expect("abort should delete the upload after UploadPart releases the lock");
    }

+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn cancelled_put_object_part_keeps_upload_lock_through_commit() {
+        for pause in [
+            MultipartCommitPause::PutPartAfterCapacityAdmission,
+            MultipartCommitPause::PutPartAfterRename,
+        ] {
+            let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+            let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+            let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+            let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+            let bucket = "multipart-cancelled-part-commit-bucket";
+            let object = "object";
+            make_bucket_on_all(&disk_stores, bucket).await;
+            let upload = set_disks
+                .new_multipart_upload(bucket, object, &ObjectOptions::default())
+                .await
+                .expect("multipart upload should be created");
+            let upload_id = upload.upload_id;
+            let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+            signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_path.clone()));
+            let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+            let barrier = MultipartCommitBarrier::install(bucket, object, pause);
+
+            let put_store = set_disks.clone();
+            let put_upload_id = upload_id.clone();
+            let put = tokio::spawn(async move {
+                let mut reader = PutObjReader::from_vec(vec![0x45; 4096]);
+                put_store
+                    .put_object_part(bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default())
+                    .await
+            });
+            barrier.wait_until_paused().await;
+            put.abort();
+            assert!(
+                put.await
+                    .expect_err("the request waiter should remain cancellable")
+                    .is_cancelled()
+            );
+
+            let abort_store = set_disks.clone();
+            let abort_upload_id = upload_id.clone();
+            let mut abort = tokio::spawn(async move {
+                abort_store
+                    .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                    .await
+            });
+            signaling.wait_for_attempts(2).await;
+            assert!(
+                tokio::time::timeout(Duration::from_millis(100), &mut abort).await.is_err(),
+                "Abort must not acquire the upload lock while a cancelled request's commit is paused at {pause:?}"
+            );
+            barrier.release();
+            tokio::time::timeout(Duration::from_secs(10), abort)
+                .await
+                .expect("Abort should finish after the commit drains")
+                .expect("Abort task should not panic")
+                .expect("Abort should delete the committed part staging");
+            for disk in &disk_stores {
+                assert!(
+                    matches!(
+                        disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await,
+                        Err(DiskError::FileNotFound)
+                    ),
+                    "the upload directory must be absent after the commit and Abort finish"
+                );
+            }
+        }
+    }
+
    #[tokio::test(flavor = "multi_thread")]
    #[serial]
    async fn put_object_part_different_part_numbers_commit_concurrently() {
@@ -9799,6 +10404,18 @@ mod tests {
                "the post-commit crash must leave the upload listable for reclamation"
            );

+            let cleanup = set_disks
+                .lock_orphan_multipart_cleanup(bucket, object, &u_new, &ObjectOptions::default(), &[])
+                .await
+                .expect("the committed upload has no staging metadata on any disk");
+            assert!(
+                cleanup
+                    .delete(bucket, object, &ObjectOptions::default())
+                    .await
+                    .expect("recover completed upload staging")
+            );
+            drop(cleanup);
+
            // The commit rename already consumed the upload's xl.meta, so the
            // retry cannot read staging. It must still return the committed
            // object: AWS CompleteMultipartUpload is idempotent for the same
@@ -9817,21 +10434,6 @@ mod tests {
                !upload_is_listed(&set_disks, bucket, object, &u_new).await,
                "the idempotent retry must reclaim staging left behind by the crash"
            );
-
-            // Reclaim the leftover exactly as the production tail does: delete_all
-            // on the upload path (abort_multipart_upload cannot — the upload's
-            // metadata is gone, so it too answers InvalidUploadID).
-            let upload_dir = SetDisks::get_upload_id_dir(bucket, object, &u_new);
-            set_disks
-                .delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_dir)
-                .await
-                .expect("sweeping the leftover upload staging should succeed");
-            assert!(
-                !upload_is_listed(&set_disks, bucket, object, &u_new).await,
-                "reclaiming the upload must drop it from the listing"
-            );
-            let (body_after, _) = read_object(&set_disks, bucket, object).await;
-            assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object");
        }

        #[tokio::test]
@@ -9878,6 +10480,18 @@ mod tests {
                            .expect("the failed post-commit completion should release its object guard"),
                    );

+                    let cleanup = set_disks
+                        .lock_orphan_multipart_cleanup(bucket, object, &u_new, &ObjectOptions::default(), &[])
+                        .await
+                        .expect("committed staging has no upload metadata");
+                    assert!(
+                        cleanup
+                            .delete(bucket, object, &ObjectOptions::default())
+                            .await
+                            .expect("orphan recovery must only remove upload staging")
+                    );
+                    drop(cleanup);
+
                    let (body, _) = read_object(&set_disks, bucket, object).await;
                    assert_eq!(body, new, "the committed replacement must remain readable after the crash");
                    let mut receipts = 0;
@@ -5030,6 +5030,339 @@ mod tests {
        shutdown.cancel();
    }

+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn repeated_abort_recovers_exact_orphan_before_decommission_drains() {
+        use crate::storage_api_contracts::namespace::NamespaceLocking as _;
+        use sha2::Digest;
+
+        let temp_dir = tempfile::tempdir().expect("create orphan recovery store dir");
+        let (_ctx, store, shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "orphan-multipart-drain", &[4, 4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("orphan-multipart-{}", uuid::Uuid::new_v4());
+        let object = "exact-object.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        let incarnation = store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation");
+        let lifecycle_guard = store.acquire_bucket_lifecycle_read_lock(&bucket).await.expect("bucket guard");
+        let mut upload_opts = ObjectOptions {
+            expected_bucket_incarnation_id: Some(incarnation),
+            ..Default::default()
+        };
+        upload_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
+        let upload = store.pools[0]
+            .new_multipart_upload(&bucket, object, &upload_opts)
+            .await
+            .expect("create source upload");
+        drop(lifecycle_guard);
+        store
+            .abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("initial abort");
+
+        let upload_sha =
+            hex_simd::encode_to_string(sha2::Sha256::digest(format!("{bucket}/{object}").as_bytes()), hex_simd::AsciiCase::Lower);
+        let upload_uuid = crate::runtime::sources::upload_uuid_suffix(&upload.upload_id);
+        let staging_dir = |pool_index: usize, disk_index: usize| {
+            temp_dir
+                .path()
+                .join(format!("pool{pool_index}/set0/disk{disk_index}"))
+                .join(crate::disk::RUSTFS_META_MULTIPART_BUCKET)
+                .join(&upload_sha)
+                .join(&upload_uuid)
+        };
+        for disk_index in 0..3 {
+            let path = staging_dir(0, disk_index);
+            tokio::fs::create_dir_all(&path).await.expect("create late part directory");
+            tokio::fs::write(path.join("part.1"), b"late part after abort")
+                .await
+                .expect("write late part fixture");
+        }
+        mark_test_pool_decommissioning(&store, 0).await;
+        assert!(
+            store
+                .list_multipart_uploads(&bucket, "", None, None, None, 100)
+                .await
+                .expect("logical upload listing")
+                .uploads
+                .is_empty()
+        );
+        assert!(store.ensure_decommission_multipart_uploads_drained_for_test(0).await.is_err());
+        assert!(
+            store
+                .abort_multipart_upload(&bucket, "different-key", &upload.upload_id, &ObjectOptions::default())
+                .await
+                .is_err(),
+            "the same UploadID for a different key must not recover this path"
+        );
+
+        // A minority metadata copy in another pool blocks recovery everywhere.
+        let conflicting_metadata = staging_dir(1, 0).join(crate::disk::STORAGE_FORMAT_FILE);
+        tokio::fs::create_dir_all(conflicting_metadata.parent().expect("metadata parent"))
+            .await
+            .expect("create conflicting pool staging");
+        tokio::fs::write(&conflicting_metadata, b"unreadable metadata")
+            .await
+            .expect("write conflicting metadata");
+        assert!(
+            store
+                .abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+                .await
+                .is_err(),
+            "no pool may be cleaned until every pool proves absence"
+        );
+        for disk_index in 0..3 {
+            assert!(staging_dir(0, disk_index).join("part.1").exists());
+        }
+        tokio::fs::remove_file(conflicting_metadata)
+            .await
+            .expect("remove conflicting fixture");
+        let (paused, release) = crate::store::multipart::pause_orphan_multipart_cleanup_for_test(&bucket);
+        let recovery_store = store.clone();
+        let recovery_bucket = bucket.clone();
+        let recovery_upload = upload.upload_id.clone();
+        let mut recovery = tokio::spawn(async move {
+            recovery_store
+                .abort_multipart_upload(&recovery_bucket, object, &recovery_upload, &ObjectOptions::default())
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(10), async {
+            tokio::select! {
+                paused = paused => paused.expect("cleanup owner announces arrival"),
+                result = &mut recovery => panic!("recovery returned before physical cleanup: {result:?}"),
+            }
+        })
+        .await
+        .expect("recovery should reach physical cleanup");
+        recovery.abort();
+        assert!(recovery.await.expect_err("request waiter is cancellable").is_cancelled());
+        let upload_lock = store.pools[0]
+            .get_disks_by_key(object)
+            .new_ns_lock(crate::disk::RUSTFS_META_MULTIPART_BUCKET, &format!("{upload_sha}/{upload_uuid}"))
+            .await
+            .expect("same upload namespace");
+        let mut upload_writer = Box::pin(upload_lock.get_write_lock(Duration::from_secs(10)));
+        let mut bucket_writer = Box::pin(store.acquire_bucket_lifecycle_write_lock(&bucket));
+        assert!(
+            futures::poll!(&mut upload_writer).is_pending(),
+            "cancelled recovery must retain the upload lock"
+        );
+        assert!(
+            futures::poll!(&mut bucket_writer).is_pending(),
+            "cancelled recovery must retain the real bucket guard"
+        );
+        release.send(()).expect("allow physical cleanup to finish");
+        let (upload_guard, bucket_guard) =
+            tokio::time::timeout(Duration::from_secs(10), async { tokio::join!(upload_writer, bucket_writer) })
+                .await
+                .expect("both guards should release after physical cleanup");
+        drop(upload_guard.expect("upload writer resumes"));
+        drop(bucket_guard.expect("bucket writer resumes"));
+        for pool_index in 0..2 {
+            for disk_index in 0..4 {
+                assert!(!staging_dir(pool_index, disk_index).exists(), "every checked staging path must be absent");
+            }
+        }
+        store
+            .ensure_decommission_multipart_uploads_drained_for_test(0)
+            .await
+            .expect("physical decommission gate should open after recovery");
+        assert!(
+            matches!(
+                store
+                    .abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+                    .await,
+                Err(StorageError::InvalidUploadID(_, _, _))
+            ),
+            "an absent upload must not claim to have recovered anything"
+        );
+        shutdown.cancel();
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn repeated_abort_keeps_no_such_upload_while_a_disk_is_offline() {
+        use sha2::Digest;
+
+        let temp_dir = tempfile::tempdir().expect("create offline-disk recovery store dir");
+        let (_ctx, store, shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "orphan-multipart-offline", &[4, 4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("orphan-multipart-{}", uuid::Uuid::new_v4());
+        let object = "offline-object.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        let incarnation = store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation");
+        let lifecycle_guard = store.acquire_bucket_lifecycle_read_lock(&bucket).await.expect("bucket guard");
+        let mut upload_opts = ObjectOptions {
+            expected_bucket_incarnation_id: Some(incarnation),
+            ..Default::default()
+        };
+        upload_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
+        let upload = store.pools[0]
+            .new_multipart_upload(&bucket, object, &upload_opts)
+            .await
+            .expect("create source upload");
+        drop(lifecycle_guard);
+        store
+            .abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("initial abort");
+
+        let upload_sha =
+            hex_simd::encode_to_string(sha2::Sha256::digest(format!("{bucket}/{object}").as_bytes()), hex_simd::AsciiCase::Lower);
+        let upload_uuid = crate::runtime::sources::upload_uuid_suffix(&upload.upload_id);
+        let staging_dir = |disk_index: usize| {
+            temp_dir
+                .path()
+                .join(format!("pool0/set0/disk{disk_index}"))
+                .join(crate::disk::RUSTFS_META_MULTIPART_BUCKET)
+                .join(&upload_sha)
+                .join(&upload_uuid)
+        };
+        for disk_index in 0..3 {
+            let path = staging_dir(disk_index);
+            tokio::fs::create_dir_all(&path).await.expect("create late part directory");
+            tokio::fs::write(path.join("part.1"), b"late part after abort")
+                .await
+                .expect("write late part fixture");
+        }
+
+        // One offline disk means absence cannot be proven. The idempotent Abort
+        // must keep answering NoSuchUpload instead of surfacing a disk error,
+        // and must not delete anything.
+        let set = store.pools[0].get_disks_by_key(object);
+        let offline = set.disks.write().await[3].take();
+        assert!(offline.is_some(), "fixture disk must have been online");
+        assert!(
+            matches!(
+                store
+                    .abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+                    .await,
+                Err(StorageError::InvalidUploadID(_, _, _))
+            ),
+            "unprovable absence must keep NoSuchUpload while a disk is offline"
+        );
+        for disk_index in 0..3 {
+            assert!(
+                staging_dir(disk_index).join("part.1").exists(),
+                "no staging may be deleted while a disk is offline"
+            );
+        }
+
+        set.disks.write().await[3] = offline;
+        store
+            .abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("recovery resumes once every disk is observable");
+        for disk_index in 0..4 {
+            assert!(!staging_dir(disk_index).exists(), "recovered staging must be absent");
+        }
+        shutdown.cancel();
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn orphan_multipart_recovery_preserves_current_bucket_incarnation() {
+        use sha2::Digest;
+
+        let temp_dir = tempfile::tempdir().expect("create bucket lifetime recovery store dir");
+        let (_ctx, store, shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "orphan-bucket-lifetime", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("orphan-lifetime-{}", uuid::Uuid::new_v4());
+        let object = "object.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create original bucket");
+        let original_incarnation = store.bucket_incarnation_id(&bucket).await.expect("original incarnation");
+        let upload = store
+            .new_multipart_upload(&bucket, object, &ObjectOptions::default())
+            .await
+            .expect("create original upload");
+        store
+            .abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("abort original upload");
+        store
+            .delete_bucket(&bucket, &crate::storage_api_contracts::bucket::DeleteBucketOptions::default())
+            .await
+            .expect("delete original bucket");
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("recreate bucket");
+        assert_ne!(store.bucket_incarnation_id(&bucket).await.expect("new incarnation"), original_incarnation);
+        let mut reader = PutObjReader::from_vec(b"current bucket object".to_vec());
+        store
+            .put_object(&bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("write current bucket object");
+        let current_upload = store
+            .new_multipart_upload(&bucket, object, &ObjectOptions::default())
+            .await
+            .expect("create current incarnation upload for the same key");
+
+        let upload_sha =
+            hex_simd::encode_to_string(sha2::Sha256::digest(format!("{bucket}/{object}").as_bytes()), hex_simd::AsciiCase::Lower);
+        let upload_uuid = crate::runtime::sources::upload_uuid_suffix(&upload.upload_id);
+        let leftover = temp_dir
+            .path()
+            .join("pool0/set0/disk0")
+            .join(crate::disk::RUSTFS_META_MULTIPART_BUCKET)
+            .join(upload_sha)
+            .join(upload_uuid)
+            .join("part.1");
+        tokio::fs::create_dir_all(leftover.parent().expect("staging parent"))
+            .await
+            .expect("create old incarnation staging");
+        tokio::fs::write(&leftover, b"previous bucket lifetime")
+            .await
+            .expect("write previous incarnation fixture");
+        for opts in [
+            ObjectOptions {
+                expected_bucket_incarnation_id: Some(original_incarnation),
+                ..Default::default()
+            },
+            ObjectOptions {
+                expected_bucket_incarnation_id: Some(uuid::Uuid::nil()),
+                ..Default::default()
+            },
+        ] {
+            assert!(
+                store
+                    .abort_multipart_upload(&bucket, object, &upload.upload_id, &opts)
+                    .await
+                    .is_err(),
+                "a stale or nil current-bucket identity must fail closed"
+            );
+            assert!(leftover.exists(), "failed bucket lifetime proof must preserve the path");
+        }
+        store
+            .abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("exact metadata-less staging from an old incarnation is unreferenced");
+        assert!(!leftover.exists());
+        store
+            .get_multipart_info(&bucket, object, &current_upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("recovery must preserve the current incarnation's active upload");
+        let mut reader = store
+            .get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("current object must remain readable");
+        let mut body = Vec::new();
+        reader.stream.read_to_end(&mut body).await.expect("read current object");
+        assert_eq!(body, b"current bucket object");
+        shutdown.cancel();
+    }
+
    #[tokio::test]
    #[serial_test::serial(storage_class_env)]
    async fn active_multipart_upload_routes_before_faulted_suspended_source() {
@@ -18591,6 +18924,70 @@ mod tests {
        assert_eq!(body, b"new-generation");
    }

+    #[tokio::test(flavor = "multi_thread")]
+    #[serial_test::serial]
+    #[serial_test::serial(storage_class_env)]
+    async fn cancelled_ecstore_put_part_keeps_bucket_lifecycle_lock_through_commit() {
+        for pause in [
+            crate::set_disk::MultipartCommitPause::PutPartAfterCapacityAdmission,
+            crate::set_disk::MultipartCommitPause::PutPartAfterRename,
+        ] {
+            let temp_dir = tempfile::tempdir().expect("create cancelled multipart lifecycle store dir");
+            let (_ctx, store, shutdown) =
+                without_storage_class_env(build_isolated_test_store(temp_dir.path(), "cancelled-multipart-lifecycle", &[4, 4]))
+                    .await;
+            crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+            let bucket = format!("cancelled-part-lifecycle-{}", Uuid::new_v4());
+            let object = "part.bin";
+            store
+                .make_bucket(&bucket, &MakeBucketOptions::default())
+                .await
+                .expect("create bucket for multipart lifecycle cancellation");
+            let upload = store
+                .new_multipart_upload(&bucket, object, &ObjectOptions::default())
+                .await
+                .expect("create upload through the owning ECStore");
+            let barrier = crate::set_disk::MultipartCommitBarrier::install(&bucket, object, pause);
+            let put_store = Arc::clone(&store);
+            let put_bucket = bucket.clone();
+            let put_upload_id = upload.upload_id.clone();
+            let put = tokio::spawn(async move {
+                let mut reader = PutObjReader::from_vec(vec![0x45; 4096]);
+                put_store
+                    .put_object_part(&put_bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default())
+                    .await
+            });
+            barrier.wait_until_paused().await;
+            put.abort();
+            assert!(
+                put.await
+                    .expect_err("the ECStore request waiter must remain cancellable")
+                    .is_cancelled()
+            );
+
+            let mut writer = Box::pin(store.acquire_bucket_lifecycle_write_lock(&bucket));
+            assert!(
+                tokio::time::timeout(Duration::from_millis(100), &mut writer).await.is_err(),
+                "canceling UploadPart at {pause:?} must not release its bucket lifecycle owner before the commit drains"
+            );
+            barrier.release();
+            let lifecycle_guard = tokio::time::timeout(Duration::from_secs(10), writer)
+                .await
+                .expect("the bucket lifecycle writer must resume after the part commit drains")
+                .expect("acquire the bucket lifecycle write lock after cancellation");
+            drop(lifecycle_guard);
+
+            let parts = store
+                .list_object_parts(&bucket, object, &upload.upload_id, None, 100, &ObjectOptions::default())
+                .await
+                .expect("the cancelled request's owned part commit must finish successfully");
+            assert_eq!(parts.parts.len(), 1, "the commit must publish exactly the uploaded part at {pause:?}");
+            assert_eq!(parts.parts[0].part_num, 1);
+            assert_eq!(parts.parts[0].size, 4096);
+            shutdown.cancel();
+        }
+    }
+
    #[tokio::test]
    #[serial_test::serial(storage_class_env)]
    async fn pre_upgrade_multipart_upload_can_complete_and_abort_in_the_same_bucket_lifetime() {
@@ -32,6 +32,42 @@ fn data_movement_multipart_discovery_counts() -> &'static std::sync::Mutex<std::
    DATA_MOVEMENT_MULTIPART_DISCOVERY_COUNTS.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
 }

+#[cfg(test)]
+type OrphanMultipartCleanupPause = (tokio::sync::oneshot::Sender<()>, tokio::sync::oneshot::Receiver<()>);
+
+#[cfg(test)]
+static ORPHAN_MULTIPART_CLEANUP_PAUSES: std::sync::OnceLock<
+    std::sync::Mutex<std::collections::HashMap<String, OrphanMultipartCleanupPause>>,
+> = std::sync::OnceLock::new();
+
+#[cfg(test)]
+pub(crate) fn pause_orphan_multipart_cleanup_for_test(
+    bucket: &str,
+) -> (tokio::sync::oneshot::Receiver<()>, tokio::sync::oneshot::Sender<()>) {
+    let (arrived_tx, arrived_rx) = tokio::sync::oneshot::channel();
+    let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+    let previous = ORPHAN_MULTIPART_CLEANUP_PAUSES
+        .get_or_init(Default::default)
+        .lock()
+        .expect("orphan cleanup pause mutex should not poison")
+        .insert(bucket.to_owned(), (arrived_tx, release_rx));
+    assert!(previous.is_none(), "each bucket has at most one orphan cleanup pause");
+    (arrived_rx, release_tx)
+}
+
+#[cfg(test)]
+async fn pause_orphan_multipart_cleanup(bucket: &str) {
+    let pause = ORPHAN_MULTIPART_CLEANUP_PAUSES
+        .get_or_init(Default::default)
+        .lock()
+        .expect("orphan cleanup pause mutex should not poison")
+        .remove(bucket);
+    if let Some((arrived, release)) = pause {
+        let _ = arrived.send(());
+        let _ = release.await;
+    }
+}
+
 fn decommission_multipart_target_clear_pending(opts: &ObjectOptions, target: Option<&ObjectInfo>) -> Result<bool> {
    let expected_mod_time = opts.mod_time.ok_or_else(|| Error::DecommissionCapacityBlocked {
        message: "multipart cleanup cannot prove exact target absence without a modification time".to_string(),
@@ -654,7 +690,10 @@ impl ECStore {
        opts: &ObjectOptions,
    ) -> Result<PartInfo> {
        check_put_object_part_args(bucket, object, upload_id)?;
-        let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        let (mut opts, bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        if let Some(guard) = bucket_lifecycle_guard {
+            opts.bucket_lifecycle_lock_owner = Some(Arc::new(guard));
+        }
        if !self.single_pool() {
            opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
        }
@@ -750,19 +789,12 @@ impl ECStore {
        opts: &ObjectOptions,
    ) -> Result<()> {
        check_abort_multipart_args(bucket, object, upload_id)?;
-        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
-        let opts = &opts;
-
-        // TODO(backlog): defer DeleteUploadID to background for faster abort response
-
-        if self.single_pool() {
-            return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await;
-        }
+        let (opts, bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;

        for pool_idx in self.existing_multipart_pool_order().await {
            let pool = &self.pools[pool_idx];

-            let err = match pool.abort_multipart_upload(bucket, object, upload_id, opts).await {
+            let err = match pool.abort_multipart_upload(bucket, object, upload_id, &opts).await {
                Ok(_) => return Ok(()),
                Err(err) => {
                    //
@@ -775,7 +807,92 @@ impl ECStore {
            }
        }

-        Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
+        self.recover_orphan_multipart_upload(bucket, object, upload_id, opts, bucket_lifecycle_guard)
+            .await
+    }
+
+    async fn recover_orphan_multipart_upload(
+        &self,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        opts: ObjectOptions,
+        bucket_lifecycle_guard: Option<rustfs_lock::NamespaceLockGuard>,
+    ) -> Result<()> {
+        let invalid_upload = || StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned());
+        if self.ctx.lock_manager().is_disabled() {
+            return Err(invalid_upload());
+        }
+        let Some(bucket_lifecycle_guard) = bucket_lifecycle_guard else {
+            // A signal borrowed from an enclosing operation cannot keep its lock
+            // alive when the request waiter disappears during physical cleanup.
+            return Err(invalid_upload());
+        };
+        let Some(expected_incarnation) = opts.expected_bucket_incarnation_id.filter(|id| !id.is_nil()) else {
+            return Err(invalid_upload());
+        };
+        // Recovery is a side path of an idempotent Abort. Until the first
+        // deletion is issued, anything that stops staging absence from being
+        // proven (an offline disk, lock contention, partially overlapping lock
+        // domains, unreadable bucket metadata) keeps the upload reported as
+        // gone, exactly as before recovery existed; only failures inside the
+        // deletion task are request failures, because a retry can finish them.
+        let (metadata, persisted) =
+            match crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await {
+                Ok(loaded) => loaded,
+                Err(error) => {
+                    debug!(bucket, object, upload_id, %error, "orphan multipart cleanup skipped: bucket metadata unavailable");
+                    return Err(invalid_upload());
+                }
+            };
+        if !persisted
+            || !metadata.bucket_creation_committed
+            || !metadata.bucket_incarnation_sidecar
+            || metadata.bucket_incarnation_id != expected_incarnation
+        {
+            return Err(invalid_upload());
+        }
+
+        // Lock order: bucket lifecycle, then upload locks in pool-index order.
+        // Prove metadata absence across every pool before deleting any staging.
+        let mut cleanup_guards = Vec::with_capacity(self.pools.len());
+        for pool in &self.pools {
+            let guard = match pool
+                .get_disks_by_key(object)
+                .lock_orphan_multipart_cleanup(bucket, object, upload_id, &opts, &cleanup_guards)
+                .await
+            {
+                Ok(guard) => guard,
+                Err(error) => {
+                    if !is_err_invalid_upload_id(&error) {
+                        debug!(bucket, object, upload_id, %error, "orphan multipart cleanup skipped: staging absence not proven");
+                    }
+                    return Err(invalid_upload());
+                }
+            };
+            cleanup_guards.push(guard);
+        }
+        let bucket = bucket.to_owned();
+        let object = object.to_owned();
+        let upload_id = upload_id.to_owned();
+        tokio::spawn(async move {
+            // Own both the real bucket guard and all upload guards until every
+            // disk deletion settles, including after cancellation of the waiter.
+            let _bucket_lifecycle_guard = bucket_lifecycle_guard;
+            #[cfg(test)]
+            pause_orphan_multipart_cleanup(&bucket).await;
+            let mut recovered = false;
+            for guard in &cleanup_guards {
+                recovered |= guard.delete(&bucket, &object, &opts).await?;
+            }
+            if recovered {
+                Ok(())
+            } else {
+                Err(StorageError::InvalidUploadID(bucket, object, upload_id))
+            }
+        })
+        .await
+        .map_err(|error| Error::other_with_context("orphan multipart cleanup task failed", error))?
    }

    pub(crate) async fn abort_multipart_upload_for_data_movement(
@@ -1873,8 +1873,10 @@ async fn root_recovery_legacy_cluster_requires_a_fresh_generation_bound_task() {

    let restarted = recovery_manager(vec![disk]);
    restarted.replay_root_heals().await.expect("replay legacy root record");
-    let queued = restarted.heal_queue.lock().await.requests().cloned().collect::<Vec<_>>();
-    assert!(queued.is_empty(), "unfenced legacy work must not bind to a later bucket generation");
+    assert!(
+        restarted.heal_queue.lock().await.requests().next().is_none(),
+        "unfenced legacy work must not bind to a later bucket generation"
+    );
    assert!(
        restarted
            .root_recovery
@@ -120,8 +120,11 @@ pub fn object_lock_put_integrity(
 /// stored bytes differ from what the source sent — e.g. a target that does not
 /// decode `aws-chunked` framing stores the frames verbatim and returns their
 /// ETag. Only a plain single-part MD5 ETag on both sides is decidable; a
-/// multipart or opaque (encrypted) ETag, or a withheld replica ETag, returns
-/// `false` because no corruption can be concluded from it.
+/// multipart ETag, or a withheld replica ETag, returns `false` because no
+/// corruption can be concluded from it. An encrypted source's keyed
+/// fingerprint is also 32 hex characters and cannot be told apart from an MD5
+/// here, so callers must exempt encrypted sources before asking (the ecstore
+/// resyncer does so through its source-encryption classifier).
 pub fn single_part_replica_etag_mismatch(source_etag: Option<&str>, replica_etag: Option<&str>) -> bool {
    let Some(source) = source_etag.map(trim_etag) else {
        return false;
@@ -2740,6 +2740,7 @@ where
            Err(e) => {
                match classify_scanner_leader_lock_failure(&e) {
                    ScannerLeaderLockFailure::Contended => {
+                        clear_scanner_transient_cycle_recovery_status();
                        record_scanner_cycle_schedule_role("follower");
                        record_scanner_leader_lock_state("contended");
                        global_metrics().record_scanner_leader_liveness("contended", false, "").await;
@@ -353,6 +353,17 @@ fn publish_scanner_cleanup_failure(reason: String, expected: u64) {
    let _ = publish_scanner_cleanup_status(status, Some(expected));
 }

+pub(super) fn clear_scanner_transient_cycle_recovery_status() {
+    // Snapshot the version before reading status so any concurrent publish rejects the CAS below.
+    let version = scanner_cleanup_status_version();
+    let status = scanner_cycle_recovery_status();
+    // A follower can discard its failed leadership observation. Durable recovery
+    // markers are revalidated before any later leadership claim.
+    if status.classification.is_none() && matches!(status.state.as_str(), "transient" | "paused") {
+        let _ = publish_scanner_cleanup_status(recovery_status("healthy", None, false), Some(version));
+    }
+}
+
 pub(super) fn record_scanner_usage_floor_failure(reason: String) {
    let previous = scanner_cycle_recovery_status();
    let same_failure = previous.classification.as_deref() == Some(USAGE_FLOOR_LOAD_FAILED);
@@ -81,6 +81,159 @@ async fn persisted_state(store: &Arc<ECStore>) -> Vec<(Option<Vec<u8>>, DataUsag
    state
 }

+async fn seed_transient_cycle_read_failure() -> ScannerCycleRecoveryStatus {
+    let store = Arc::new(MemoryConfigStore::default());
+    let error = EcstoreError::InsufficientReadQuorum(RUSTFS_META_BUCKET.to_string(), DATA_USAGE_BLOOM_NAME_PATH.clone());
+    let expected_error = format!("Scanner error: failed to inspect scanner cycle state: {error}");
+    store
+        .read_errors
+        .lock()
+        .await
+        .insert(memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()), error);
+    let ScannerCycleStateStartup::Transient(error) = load_scanner_cycle_state_for_startup(store).await else {
+        panic!("a quorum outage must produce transient cycle recovery");
+    };
+    assert_eq!(error.to_string(), expected_error);
+    let status = scanner_cycle_recovery_status();
+    assert_eq!(status.state, "transient");
+    assert!(status.retryable);
+    assert_eq!(status.classification, None);
+    status
+}
+
+#[tokio::test]
+#[serial]
+async fn issue_8297_contended_follower_clears_only_transient_cycle_recovery() {
+    temp_env::async_with_vars([("RUSTFS_LOCK_ACQUIRE_TIMEOUT", Some("1"))], async {
+        let (_dir, store) = setup_scanner_cycle_store().await;
+        seed_cleanup(&store, "cleanup-pending").await;
+        let before = persisted_state(&store).await;
+        let lock = store
+            .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock")
+            .await
+            .expect("leader lock");
+        let guard = lock
+            .get_write_lock_quiet(Duration::from_secs(1))
+            .await
+            .expect("hold live leader");
+        let transient = seed_transient_cycle_read_failure().await;
+        assert!(!record_scanner_cycle_recovery_retry(MAX_SCANNER_CYCLE_RECOVERY_RETRIES));
+        let paused = scanner_cycle_recovery_status();
+        assert_eq!(paused.state, "paused");
+        let mut cases = vec![(transient, true), (paused, true)];
+        for (state, classification, retryable) in [
+            ("recovery-required", None, false),
+            ("blocked", None, false),
+            ("blocked", Some("corrupt"), false),
+            ("cleanup-pending", None, true),
+            ("cleanup-pending", Some("corrupt"), true),
+            ("usage_floor_load_failed", Some("usage_floor_load_failed"), true),
+            ("usage_floor_recovery_pending", Some("legacy_empty_usage_floor"), true),
+            ("paused", Some("usage_floor_load_failed"), false),
+            ("cache_cycle_ahead", Some("cache_cycle_ahead"), true),
+        ] {
+            cases.push((
+                ScannerCycleRecoveryStatus {
+                    state: state.to_string(),
+                    classification: classification.map(str::to_string),
+                    retry_count: 5,
+                    max_retries: MAX_SCANNER_CYCLE_RECOVERY_RETRIES,
+                    retryable,
+                    reason: Some("preserve recovery state".to_string()),
+                    ..Default::default()
+                },
+                false,
+            ));
+        }
+        for (status, clear) in cases {
+            let expected = serde_json::to_value(&status).expect("expected recovery status");
+            publish_scanner_cleanup_status(status, None).expect("seed local recovery status");
+            tokio::time::timeout(
+                Duration::from_secs(30),
+                run_data_scanner_with_maintenance_state(
+                    CancellationToken::new(),
+                    store.clone(),
+                    ScannerMaintenanceFeatures::default(),
+                    None,
+                ),
+            )
+            .await
+            .expect("contended scanner iteration must finish")
+            .expect("lock contention is a normal follower outcome");
+            assert_eq!(scanner_cycle_schedule_status().execution_role, "follower");
+            let actual = scanner_cycle_recovery_status();
+            if clear {
+                assert_eq!(actual.state, "healthy", "stale follower recovery: {expected}");
+                assert_eq!(actual.retry_count, 0);
+                assert!(!actual.retryable);
+            } else {
+                assert_eq!(serde_json::to_value(actual).expect("actual recovery status"), expected);
+            }
+            assert!(!guard.is_lock_lost(), "follower must not revoke the leader");
+            assert_eq!(persisted_state(&store).await, before, "follower must not rewrite durable recovery state");
+        }
+    })
+    .await;
+}
+
+#[tokio::test]
+#[serial]
+async fn issue_8297_supervisor_recovers_paused_follower_without_restart() {
+    temp_env::async_with_vars(
+        [
+            ("RUSTFS_LOCK_ACQUIRE_TIMEOUT", Some("1")),
+            (ENV_SCANNER_START_DELAY_SECS, Some("0")),
+            (ENV_SCANNER_CYCLE, Some("1")),
+        ],
+        async {
+            let (_dir, store) = setup_scanner_cycle_store().await;
+            seed_cleanup(&store, "cleanup-pending").await;
+            let before = persisted_state(&store).await;
+            let lock = store
+                .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock")
+                .await
+                .expect("leader lock");
+            let guard = lock
+                .get_write_lock_quiet(Duration::from_secs(1))
+                .await
+                .expect("hold live leader");
+            for attempt in 1..=MAX_SCANNER_CYCLE_RECOVERY_RETRIES {
+                seed_transient_cycle_read_failure().await;
+                assert_eq!(record_scanner_cycle_recovery_retry(attempt), attempt < MAX_SCANNER_CYCLE_RECOVERY_RETRIES);
+            }
+            assert_eq!(scanner_cycle_recovery_status().state, "paused");
+            assert!(!crate::scanner_runtime_initialized(), "test must own the scanner runtime");
+            let ctx = CancellationToken::new();
+            let _cancel_on_drop = ctx.clone().drop_guard();
+            init_data_scanner_with_storage(ctx.clone(), store.clone()).await;
+            assert!(crate::scanner_runtime_initialized());
+            let recovered = tokio::time::timeout(Duration::from_secs(30), async {
+                while scanner_cycle_recovery_status().state != "healthy" {
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+            })
+            .await;
+            ctx.cancel();
+            tokio::time::timeout(Duration::from_secs(15), async {
+                while crate::scanner_runtime_initialized() {
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+            })
+            .await
+            .expect("cancelled scanner supervisor must stop");
+            recovered.expect("follower must recover without another five-minute pause or process restart");
+            assert_eq!(scanner_cycle_schedule_status().execution_role, "follower");
+            let status = scanner_cycle_recovery_status();
+            assert_eq!(status.retry_count, 0);
+            assert!(!status.retryable);
+            assert!(!guard.is_lock_lost(), "supervisor must not revoke the leader");
+            assert_eq!(persisted_state(&store).await, before, "supervisor must leave durable recovery untouched");
+        },
+    )
+    .await;
+    crate::runtime_config::refresh_scanner_runtime_config_for_tests();
+}
+
 async fn run_disabled_startup(ctx: CancellationToken, store: Arc<ECStore>) {
    let initialized_before = crate::scanner_runtime_initialized();
    let cleanup = init_scanner_with_recovery(ctx, store, false).await;
@@ -24,6 +24,17 @@ Clients that require an ETag for these older or multipart objects must issue an
 authorized HeadObject request, including the SSE-C key when required. GET,
 HEAD, and conditional requests continue to use the object's stored ETag.

+## Replication
+
+Bucket and site replication forward the stored ETag on `x-rustfs-source-etag`
+(with its `x-minio-source-*` twin). A target that adopts it (RustFS, MinIO)
+reports the fingerprint for the replica; a target that computes its own ETag
+(AWS S3 and most compatible stores) reports the MD5 of the plaintext it
+received, which never equals the fingerprint. The post-PUT replica ETag audit
+therefore skips encrypted sources, and the checksum header on the replication
+PUT remains their integrity check. See
+`docs/operations/replication-outbound-transport.md`.
+
 ## Existing data and migration

 The upgrade does not rewrite existing object metadata or erase fingerprints
@@ -9,7 +9,7 @@
 - For a single-part object, the checksum the source object was uploaded with, forwarded as its `x-amz-checksum-<algorithm>` header (the value the source verified on upload). A multipart replica is rebuilt through CreateMultipartUpload/UploadPart and carries no object-level checksum header. Managed-SSE objects forward none.
 - On a PUT that carries Object Lock parameters and no forwarded checksum: `Content-MD5` derived from the source ETag, or an SDK CRC32 checksum when the ETag is not the MD5 of the wire bytes (rustfs#7082).
 - The source ETag, mtime and version id on `x-rustfs-source-*` headers (with `x-minio-source-*` twins), and the Object Lock mode, retain-until date and legal hold of the source version when present.
- After the PUT, the target's ETag is compared with the source ETag when both are plain single-part MD5s; a mismatch fails the replication instead of reporting a corrupted replica as COMPLETED.
+- After the PUT, the target's ETag is compared with the source ETag when both are plain single-part MD5s and the source is unencrypted; a mismatch fails the replication instead of reporting a corrupted replica as COMPLETED. A managed-SSE (SSE-S3 / SSE-KMS) or SSE-C source is not audited this way: its stored ETag is a keyed fingerprint (new writes, see `docs/operations/encrypted-etags.md`) or a legacy plaintext MD5, and the target re-encrypts what it receives, so the two ETags are never comparable. The checksum header on the replication PUT remains the integrity check for those objects.

 ## Target classes and their known requirements

@@ -130,6 +130,7 @@ The request must be authenticated with an admin identity that has `ServerInfoAdm
 |---|---|
 | `runtime_config` | Effective runtime controls and their value sources. |
 | `cycle_schedule` | Current effective cycle interval and clean-idle backoff state. |
+| `cycle_recovery` | This node's cycle-state recovery observation, retry budget, and reason. |
 | `metrics` | Scanner work, pressure, checkpoint, lifecycle, replication, heal, bitrot, and alert counters. |
 | `data_movement_pause` | Global-pause policy, current movement reason, operation epoch, start time, duration, and estimated movement work items. |
 | `pause_backlog` | Replicated durable pause ledger, post-pause catch-up phase, rate window, retry state, thresholds, and active alert reasons. |
@@ -190,6 +191,12 @@ Shared compacted accumulators defer periodic emission until the owning subtree c

 A confirmed periodic save does not prove restart resumability. Adoption still validates bucket/set identity and, for non-empty caches, connected cache scope. Skipping completed subtrees additionally requires a validated coverage frontier; raw enumeration state alone does not prove completed subtree coverage.

+### Cycle recovery on followers
+
+Cycle recovery status is local to each node. A transient cycle-state read failure can exhaust the short retry budget and enter `paused`; the enabled scanner then retries leadership every five minutes or when recovery is explicitly woken. If the next lock attempt is contended or times out waiting for `leader.lock`, the node records a follower iteration and clears only its unclassified `transient` or `paused` observation. Normal follower retry cadence resumes without restarting the process.
+
+This clears local retry bookkeeping, not persisted cycle state or recovery markers. Classified recovery states, cleanup-pending resets, and usage-floor failures remain unchanged. Any later leadership claim revalidates durable cycle state before scanning. A healthy follower recovery status does not prove that leader-dispatched scans are progressing; inspect the work counters separately.
+
 ## Usage State Reset

 The supported break-glass route for rebuilding scanner usage state is `POST /v3/scanner/usage-state/reset` with body `{"mode":"full-rebuild"}`, authenticated as an admin identity holding `ConfigUpdateAdminAction` (route registered in `rustfs/src/admin/route_registration_test.rs`). Use it only after the scanner status shows a usage-floor load failure, a conflicting persisted usage floor, or an operator decision to discard the durable usage baseline and rebuild it from a full scanner pass.
@@ -42,12 +42,9 @@ use s3s::{
    s3_error,
 };
 use serde::{Deserialize, Serialize};
-use std::{
-    future::Future,
-    sync::Arc,
-    time::{Duration, Instant},
-};
+use std::{future::Future, sync::Arc, time::Duration};
 use time::OffsetDateTime;
+use tokio::time::Instant;
 use tracing::{error, info, warn};

 const LOG_COMPONENT_ADMIN: &str = "admin";
@@ -128,6 +125,13 @@ fn is_rebalance_start_fleet_proof_retryable(err: &StorageError) -> bool {
        && err.to_string().contains(REBALANCE_START_FLEET_PROOF_MARKER)
 }

+fn rebalance_init_failure_error(err: &StorageError) -> S3Error {
+    if is_rebalance_start_fleet_proof_retryable(err) {
+        return rebalance_fleet_proof_unavailable_error();
+    }
+    s3_error!(InternalError, "failed to initialize rebalance: {}", err)
+}
+
 fn rebalance_start_failure_error(start_err: &StorageError, rollback_result: &Result<(), String>) -> S3Error {
    if crate::storage_api::capacity::is_pool_activation_fleet_proof_error(start_err) {
        if rollback_result.is_ok() {
@@ -611,7 +615,20 @@ impl Operation for RebalanceStart {

        let buckets: Vec<String> = bucket_infos.into_iter().map(|bucket| bucket.name).collect();

-        let id = match store.init_rebalance_start(buckets).await {
+        let fleet_proof_deadline = Instant::now() + REBALANCE_START_FLEET_PROOF_RETRY_BUDGET;
+        // A missing proof fails before init publishes metadata. Retrying the full
+        // operation reacquires its gates and revalidates competing starts.
+        let id = match retry_rebalance_start_operation(
+            "init_rebalance_start",
+            "",
+            &request_id,
+            &actor,
+            &remote_addr,
+            fleet_proof_deadline,
+            || store.init_rebalance_start(buckets.clone()),
+        )
+        .await
+        {
            Ok(id) => id,
            Err(StorageError::DecommissionAlreadyRunning) => {
                log_rebalance_request_rejected("start", "decommission_in_progress", &request_id, &actor, &remote_addr);
@@ -622,7 +639,7 @@ impl Operation for RebalanceStart {
                return Err(s3_error!(OperationAborted, "rebalance is already in progress"));
            }
            Err(e) => {
-                return Err(s3_error!(InternalError, "failed to initialize rebalance: {}", e));
+                return Err(rebalance_init_failure_error(&e));
            }
        };

@@ -639,7 +656,6 @@ impl Operation for RebalanceStart {
            "admin rebalance state"
        );
        let notification_sys = current_notification_system();
-        let fleet_proof_deadline = Instant::now() + REBALANCE_START_FLEET_PROOF_RETRY_BUDGET;
        for step in rebalance_start_steps(notification_sys.is_some()) {
            match step {
                RebalanceStartStep::PropagateFence => {
@@ -1170,10 +1186,10 @@ mod rebalance_handler_tests {
        Body, HeaderMap, Method, Operation, Params, RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStart,
        RebalanceStartStep, RebalanceStatus, RebalanceStop, RebalanceStopPropagationStatus, S3ErrorCode, S3Request, StatusCode,
        Uri, build_rebalance_admin_status, build_rebalance_pool_statuses, build_rebalance_stop_propagation_status,
-        is_rebalance_start_fleet_proof_retryable, rebalance_pool_used, rebalance_query_present, rebalance_remaining_buckets,
-        rebalance_rollback_failure_message, rebalance_rollback_stop_failure_message, rebalance_start_failure_error,
-        rebalance_start_rollback_error, rebalance_start_steps, rebalance_stop_target_id, rebalance_used_pct,
-        retry_rebalance_start_operation, rollback_result_label, stop_rebalance_admission_first,
+        is_rebalance_start_fleet_proof_retryable, rebalance_init_failure_error, rebalance_pool_used, rebalance_query_present,
+        rebalance_remaining_buckets, rebalance_rollback_failure_message, rebalance_rollback_stop_failure_message,
+        rebalance_start_failure_error, rebalance_start_rollback_error, rebalance_start_steps, rebalance_stop_target_id,
+        rebalance_used_pct, retry_rebalance_start_operation, rollback_result_label, stop_rebalance_admission_first,
    };
    use crate::admin::storage_api::error::StorageError;
    use crate::admin::storage_api::rebalance::{
@@ -1533,7 +1549,7 @@ mod rebalance_handler_tests {
    }

    #[tokio::test(start_paused = true)]
-    async fn test_rebalance_start_retry_waits_for_fleet_proof() {
+    async fn test_rebalance_init_retry_waits_for_fleet_proof() {
        use std::sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
@@ -1544,12 +1560,12 @@ mod rebalance_handler_tests {
            let attempts = Arc::clone(&attempts);
            async move {
                retry_rebalance_start_operation(
-                    "load_rebalance_meta(start=true)",
-                    "rebalance-id",
+                    "init_rebalance_start",
+                    "",
                    "request-id",
                    "actor",
                    "remote",
-                    std::time::Instant::now() + super::REBALANCE_START_FLEET_PROOF_RETRY_BUDGET,
+                    super::Instant::now() + super::REBALANCE_START_FLEET_PROOF_RETRY_BUDGET,
                    move || {
                        let attempts = Arc::clone(&attempts);
                        async move {
@@ -1557,7 +1573,7 @@ mod rebalance_handler_tests {
                            if attempt == 0 {
                                Err(StorageError::other(super::REBALANCE_START_FLEET_PROOF_MARKER))
                            } else {
-                                Ok(())
+                                Ok("initialized-once")
                            }
                        }
                    },
@@ -1570,12 +1586,78 @@ mod rebalance_handler_tests {
        assert_eq!(attempts.load(Ordering::SeqCst), 1);

        tokio::time::advance(super::REBALANCE_START_FLEET_PROOF_RETRY_DELAY).await;
-        task.await
+        let id = task
+            .await
            .expect("retry task should not panic")
            .expect("fleet proof retry should eventually succeed");
+        assert_eq!(id, "initialized-once");
        assert_eq!(attempts.load(Ordering::SeqCst), 2);
    }

+    #[tokio::test(start_paused = true)]
+    async fn test_rebalance_init_retry_exhaustion_returns_503_with_retry_after() {
+        let mut attempts = 0;
+        let started = super::Instant::now();
+        let deadline = started + super::REBALANCE_START_FLEET_PROOF_RETRY_BUDGET;
+        let err = retry_rebalance_start_operation::<(), _, _>(
+            "init_rebalance_start",
+            "",
+            "request-id",
+            "actor",
+            "remote",
+            deadline,
+            || {
+                attempts += 1;
+                std::future::ready(Err(StorageError::other(super::REBALANCE_START_FLEET_PROOF_MARKER)))
+            },
+        )
+        .await
+        .expect_err("a missing proof must exhaust the bounded retry budget");
+
+        assert_eq!(super::Instant::now() - started, super::REBALANCE_START_FLEET_PROOF_RETRY_BUDGET);
+        assert_eq!(attempts, 7, "one immediate attempt plus six bounded readiness retries");
+        let response = rebalance_init_failure_error(&err);
+        assert_eq!(response.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
+        assert_eq!(
+            response
+                .headers()
+                .and_then(|headers| headers.get(http::header::RETRY_AFTER))
+                .and_then(|value| value.to_str().ok()),
+            Some(super::REBALANCE_START_RETRY_AFTER_SECS)
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn test_rebalance_init_does_not_retry_other_failures() {
+        for source in [
+            StorageError::other("disk read failed"),
+            StorageError::other("pool activation fleet capability proof expired before commit"),
+            StorageError::RebalanceAlreadyRunning,
+        ] {
+            let original = source.to_string();
+            let mut source = Some(source);
+            let started = super::Instant::now();
+            let err = retry_rebalance_start_operation::<(), _, _>(
+                "init_rebalance_start",
+                "",
+                "request-id",
+                "actor",
+                "remote",
+                started + super::REBALANCE_START_FLEET_PROOF_RETRY_BUDGET,
+                || std::future::ready(Err(source.take().expect("non-readiness errors must not be retried"))),
+            )
+            .await
+            .expect_err("non-readiness errors must be returned unchanged");
+            assert_eq!(err.to_string(), original);
+            assert_eq!(super::Instant::now(), started, "non-readiness failures must not wait");
+            if !matches!(err, StorageError::RebalanceAlreadyRunning) {
+                let response = rebalance_init_failure_error(&err);
+                assert_eq!(response.code(), &S3ErrorCode::InternalError);
+                assert!(response.message().is_some_and(|message| message.contains(&original)));
+            }
+        }
+    }
+
    #[test]
    fn test_rebalance_rollback_stop_failure_message_lists_failures() {
        let failures = vec![
@@ -23,6 +23,13 @@ const MAX_VERSIONS_EXCEEDED_CODE: &str = "MaxVersionsExceeded";
 const MAX_VERSIONS_EXCEEDED_MESSAGE: &str = "You've exceeded the limit on the number of versions you can create on this object";
 const SLOW_DOWN_READ_CODE: &str = "SlowDownRead";
 const SLOW_DOWN_READ_MESSAGE: &str = "Resource requested is unreadable, please reduce your request rate";
+/// Public message for `StorageError::UnsupportedDisk` on a request path. The
+/// only request-time producer is the delete-transaction protocol when a
+/// participant answers Unimplemented (a peer still on a release without the
+/// recovery RPC, or a protocol-version mismatch); the variant's Display text
+/// ("drive does not support O_DIRECT") describes the startup drive probe and
+/// would send an operator mid-upgrade after the wrong cause.
+const UNSUPPORTED_PARTICIPANT_MESSAGE: &str = "A storage participant does not support this operation's protocol version. If the cluster is being upgraded, finish upgrading every node and retry.";

 /// S3 error code for a request that names a KMS key the KMS does not hold.
 pub const KMS_KEY_NOT_FOUND_ERROR_CODE: &str = "KMS.NotFoundException";
@@ -739,6 +746,9 @@ impl From<StorageError> for ApiError {
            StorageError::ObjectNameInvalid(_, _) => S3ErrorCode::InvalidArgument,
            StorageError::BucketExists(_) => S3ErrorCode::BucketAlreadyOwnedByYou,
            StorageError::StorageFull => S3ErrorCode::ServiceUnavailable,
+            // Retryable on an upgraded node (or once the cluster upgrade
+            // completes); see UNSUPPORTED_PARTICIPANT_MESSAGE.
+            StorageError::UnsupportedDisk => S3ErrorCode::ServiceUnavailable,
            StorageError::SlowDown => S3ErrorCode::SlowDown,
            StorageError::FaultyDisk
            | StorageError::FaultyRemoteDisk
@@ -782,6 +792,8 @@ impl From<StorageError> for ApiError {

        let message = if matches!(&err, StorageError::QuotaExceeded { .. }) {
            err.to_string()
+        } else if matches!(&err, StorageError::UnsupportedDisk) {
+            UNSUPPORTED_PARTICIPANT_MESSAGE.to_string()
        } else if matches!(&err, StorageError::MaxVersionsExceeded)
            || (code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)))
        {
@@ -960,6 +972,40 @@ mod tests {
        assert!(!body.contains("private-key"));
    }

+    #[tokio::test]
+    async fn unsupported_participant_is_retryable_with_an_actionable_message() {
+        use http_body_util::BodyExt as _;
+
+        let error = S3Error::from(ApiError::from(StorageError::UnsupportedDisk));
+        assert_eq!(error.code(), &S3ErrorCode::ServiceUnavailable);
+        assert_eq!(error.message(), Some(UNSUPPORTED_PARTICIPANT_MESSAGE));
+        assert!(matches!(
+            error.source().and_then(|source| source.downcast_ref::<StorageError>()),
+            Some(StorageError::UnsupportedDisk)
+        ));
+
+        let response = error
+            .to_http_response()
+            .expect("unsupported participant error must serialize");
+        assert_eq!(response.status(), http::StatusCode::SERVICE_UNAVAILABLE);
+        assert!(
+            !response.headers().contains_key(http::header::RETRY_AFTER),
+            "retrying the same node does not help until it or its peers are upgraded"
+        );
+        let body = response
+            .into_body()
+            .collect()
+            .await
+            .expect("error XML must be readable")
+            .to_bytes();
+        let xml = std::str::from_utf8(&body).expect("error XML must be UTF-8");
+        assert!(xml.contains("<Code>ServiceUnavailable</Code>"));
+        assert!(
+            !xml.contains("O_DIRECT"),
+            "the drive-probe Display text must not leak as the public message"
+        );
+    }
+
    #[test]
    fn lock_timeout_response_preserves_wrapped_source_and_requires_503() {
        for code in [S3ErrorCode::ServiceUnavailable, S3ErrorCode::RequestTimeout] {
@@ -1853,6 +1899,7 @@ mod tests {
            ),
            (StorageError::BucketExists("test".into()), S3ErrorCode::BucketAlreadyOwnedByYou),
            (StorageError::StorageFull, S3ErrorCode::ServiceUnavailable),
+            (StorageError::UnsupportedDisk, S3ErrorCode::ServiceUnavailable),
            (StorageError::SlowDown, S3ErrorCode::SlowDown),
            (StorageError::FaultyDisk, S3ErrorCode::ServiceUnavailable),
            (StorageError::FaultyRemoteDisk, S3ErrorCode::ServiceUnavailable),
Author	SHA1	Message	Date
overtrue	331fb89628	chore(release): prepare 1.0.3	2026-10-17 15:57:35 +08:00
Chris	2b243d8659	fix(scanner): clear transient cycle recovery on follower lock contention (1.0.2 backport) (#8596 ) fix(scanner): clear transient cycle recovery on follower lock contention (#8566) (cherry picked from commit `606d8a3a8a`) Co-authored-by: Ramakrishna Chilaka <49393831+RamakrishnaChilaka@users.noreply.github.com>	2026-10-16 04:05:23 +08:00
Chris	c52570d385	fix(rebalance): wait for readiness and admit peer workers (1.0.2 backport) (#8593 ) fix(rebalance): wait for readiness and admit peer workers (#8570) * fix(rebalance): avoid contention when peers join the same run * fix(rebalance): wait for fleet proof during initialization * test(rebalance): check contextualized competing run errors * fix(rebalance): retain typed worker admission lock errors * test(rebalance): retry the init readiness 503 and pin the retry error shape (#8586) The e2e harness only retried the pre-fix 500 carrying the fleet-proof marker; the init stage now answers 503 + Retry-After after its own readiness budget, so start_rebalance treats that as the same wait. The missing-proof regression also asserts the typed error shape that the admin handler's retry predicate depends on, not only the rendered message. --------- (cherry picked from commit `7682a12b58`) Co-authored-by: cxymds <cxymds@gmail.com>	2026-10-15 18:11:29 +00:00
Chris	e0b1635abf	fix(multipart): retain commit guards and reclaim orphan staging (1.0.2 backport) (#8592 ) fix(multipart): retain commit guards across request cancellation (#8571) * fix(multipart): retain commit guards across request cancellation * fix(multipart): preserve commit task error context * fix(multipart): reclaim verified orphan upload staging (#8572) * fix(multipart): reclaim verified orphan upload staging * fix(multipart): share orphan guards across matching lock domains * fix(multipart): preserve orphan cleanup task error context * fix(ecstore): box guarded delete transition identity * fix(multipart): keep NoSuchUpload when orphan recovery cannot prove absence (#8587) Recovery is a side path of an idempotent Abort. An offline disk, lock contention, partially overlapping lock domains or unreadable bucket metadata all stop the staging-absence proof before any deletion; those used to propagate as 503/500, where the base answered NoSuchUpload. They now keep that answer and delete nothing; only failures inside the deletion task remain request failures, because a retry can finish them. --------- (cherry picked from commit `65f30bd5c1`) Co-authored-by: cxymds <cxymds@gmail.com>	2026-10-15 18:00:45 +00:00
Chris	f0f2b0cf48	fix(replication): skip the replica ETag audit for encrypted sources (1.0.2 backport) (#8580 ) Backport of #8573 (main commit `7a32ca2adb`) to release/1.0.2. The five touched files were identical between release/1.0.2 and the main base the fix was written against, so this applied without conflicts. New encrypted writes store a keyed HMAC fingerprint as the object ETag (docs/operations/encrypted-etags.md). Replication forwards that ETag on x-rustfs-source-etag, but a target that computes its own ETag (AWS S3 and most compatible stores) answers the plaintext PUT with the MD5 of what it received. Both values are 32 hex characters, so verify_single_part_replica treated the fingerprint as a comparable MD5 and failed every single-part SSE-S3 / SSE-KMS replica with "replica etag mismatch"; MRF and resync then retried the object forever. RustFS and MinIO targets adopt the source ETag and were unaffected. Exempt sources whose stored ETag cannot describe the wire bytes: any managed-SSE or SSE-C source, decided by the existing replication source-encryption classifier. The checksum header on the replication PUT remains the integrity check for those objects; plaintext sources keep the audit unchanged. Claude-Session: https://claude.ai/code/session_01CezCdjE9AdShFUi9tNk797 (cherry picked from commit `7a32ca2adb`) Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>	2026-10-15 22:47:37 +08:00
Chris	0882e99c32	fix(s3): make an unsupported delete participant retryable and explain it (1.0.2 backport) (#8577 ) Backport of #8574 (main commit `d7156290d5`) to release/1.0.2. A user-bucket delete fans out through the delete-transaction RPC. When a participant answers Unimplemented (a peer still running a release without that RPC, or a protocol-version mismatch) the coordinator reports StorageError::UnsupportedDisk, which had no S3 mapping: clients received a 500 InternalError whose message was the variant's Display text, "drive does not support O_DIRECT". That text describes the startup drive probe and sends an operator mid-upgrade after the wrong cause. Map UnsupportedDisk to 503 ServiceUnavailable with a constant message that names the protocol-version cause and the remedy (finish upgrading every node, then retry). The typed source is retained for diagnostics and no Retry-After is added, since retrying the same node does not help until the cluster upgrade completes. Conflict resolution against release/1.0.2: the DiskFull arm and the capacity-failure test from #8553 are not on this branch and were not carried over; only the UnsupportedDisk mapping, its message, the mapping table row and the new unit test are included. (cherry picked from commit `d7156290d5`) Claude-Session: https://claude.ai/code/session_01CezCdjE9AdShFUi9tNk797 Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>	2026-10-15 22:01:36 +08:00
Chris	802ac8a1bd	test(heal): avoid needless collect in legacy root recovery test (1.0.2 backport) (#8551 ) test(heal): avoid needless collect in legacy root recovery test	2026-10-14 08:22:12 -07:00