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>

crates/e2e_test/src/distributed/harness.rs

@@ -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;
         }

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

@@ -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;

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

@@ -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

crates/ecstore/src/core/pools.rs

@@ -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,

crates/ecstore/src/object_api/types.rs

@@ -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)

crates/ecstore/src/services/rebalance/control.rs

@@ -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() {

crates/ecstore/src/services/rebalance/entry.rs

@@ -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)

crates/ecstore/src/services/rebalance/runtime.rs

@@ -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);

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

@@ -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;

crates/ecstore/src/store/init.rs

@@ -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() {

crates/ecstore/src/store/multipart.rs

@@ -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(

crates/heal/src/heal/manager/tests/root_recovery.rs

@@ -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

crates/replication/src/object.rs

@@ -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;

crates/scanner/src/scanner.rs

@@ -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;

crates/scanner/src/scanner/cycle_state.rs

@@ -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);

crates/scanner/src/scanner/tests/recovery_control.rs

@@ -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;

docs/operations/encrypted-etags.md

@@ -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

docs/operations/replication-outbound-transport.md

@@ -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
 

docs/operations/scanner-runtime-controls.md

@@ -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.

rustfs/src/admin/handlers/rebalance.rs

@@ -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![

rustfs/src/error.rs

@@ -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),