test: trace rename tail stages

.config/nextest.toml

@@ -934,6 +934,12 @@ test-group = 'e2e-inline-boundaries'
 filter = 'package(e2e_test) & test(/^heal_erasure_disk_rebuild_test::/)'
 test-group = 'e2e-cluster-nightly'
 
+# The multi-pool crash fixture seeds three nodes and 36 drives. Reserve this
+# nextest run's workers so other multi-node tests do not starve its deadline.
+[[profile.e2e-full.overrides]]
+filter = 'package(e2e_test) & test(=heal_erasure_disk_rebuild_test::tests::test_cluster_root_heal_recovers_ec84_shards_across_multi_pool_after_background_target_crash)'
+threads-required = "num-test-threads"
+
 [[profile.e2e-full.overrides]]
 filter = 'package(e2e_test) & (test(/^kms::kms_vault_test::/) | test(/^kms::kms_rekey_sweep_test::/) | test(/^kms::configured_roundtrip_test::test_configured_vault_kms_admin_and_versioned_cleanup$/))'
 test-group = 'e2e-vault'

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -1700,6 +1700,15 @@ mod tests {
         );
 
         let mut cluster = RustFSTestClusterEnvironment::with_topology(topology.cluster_topology()).await?;
+        let bucket = "heal-restart-during-rebuild";
+        let mrf_fallback_scope = if !background_enabled {
+            let path = PathBuf::from(&cluster.temp_dir).join("skip-outage-mrf-fallback");
+            std::fs::write(&path, format!("{bucket}/cluster/written-while-node-down-"))?;
+            cluster.set_env("RUSTFS_E2E_SKIP_MRF_FALLBACK_PATH", path.to_string_lossy().into_owned());
+            Some(path)
+        } else {
+            None
+        };
         cluster.set_env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true");
         cluster.set_env("RUSTFS_HEAL_ENABLED", "true");
         if let Some(storage_class) = evidence_case.and_then(|case| case.storage_class_standard) {
@@ -1748,7 +1757,6 @@ mod tests {
         cluster.start_with_binary(&server_binary).await?;
         let clients = cluster.create_all_clients()?;
 
-        let bucket = "heal-restart-during-rebuild";
         clients[0].create_bucket().bucket(bucket).send().await?;
 
         let ReplacementDriveSelection {
@@ -1944,6 +1952,21 @@ mod tests {
             }
         };
 
+        if let Some(path) = &mrf_fallback_scope {
+            let observed = path.with_extension("observed");
+            let expected = format!("{bucket}/{outage_key}");
+            let deadline = Instant::now() + Duration::from_secs(30);
+            loop {
+                if tokio::fs::read_to_string(&observed).await.ok().as_deref() == Some(expected.as_str()) {
+                    break;
+                }
+                if Instant::now() >= deadline {
+                    return Err("outage repair did not reach the isolated best-effort fallback before target rejoin".into());
+                }
+                sleep(Duration::from_millis(50)).await;
+            }
+        }
+
         assert!(
             !outage_target_manifest_required
                 || target_set_peer_drives
@@ -2056,7 +2079,12 @@ mod tests {
             } else {
                 cluster_heal_is_idle(&recovered)
             };
-            if ready {
+            let pool_ready = if ready && background_enabled && !outage_write_deferred_until_rejoin {
+                all_nodes_pool_metadata_readiness(&cluster, recovery_deadline).await.0
+            } else {
+                true
+            };
+            if ready && pool_ready {
                 break;
             }
             if Instant::now() >= recovery_deadline {
@@ -2553,6 +2581,10 @@ mod tests {
             );
         }
 
+        if let Some(path) = &mrf_fallback_scope {
+            std::fs::remove_file(path)?;
+        }
+
         if let Some(cycle_end) = scanner_cycle_floor {
             wait_for_scanner_cycle_after(&cluster, cycle_end).await?;
         }

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

@@ -149,6 +149,42 @@ use tokio::sync::{RwLock, oneshot};
 use tokio::task::JoinSet;
 
 pub(in crate::set_disk) const EVENT_SET_DISK_READ: &str = "set_disk_read";
+const EVENT_SET_DISK_RENAME_PROBE: &str = "set_disk_rename_probe";
+
+fn rename_probe_enabled(bucket: &str, object: &str) -> bool {
+    static TARGET: OnceLock<Option<String>> = OnceLock::new();
+    TARGET
+        .get_or_init(|| std::env::var("RUSTFS_E2E_DIAG_RENAME_TARGET").ok())
+        .as_deref()
+        .and_then(|target| target.strip_prefix(bucket))
+        .and_then(|target| target.strip_prefix('/'))
+        == Some(object)
+}
+
+fn rename_probe(
+    state: &'static str,
+    bucket: &str,
+    object: &str,
+    transaction: &str,
+    disk_index: Option<usize>,
+    local: Option<bool>,
+    success: Option<bool>,
+) {
+    info!(
+        target: "rustfs_ecstore::rename_probe",
+        event = EVENT_SET_DISK_RENAME_PROBE,
+        component = LOG_COMPONENT_ECSTORE,
+        subsystem = LOG_SUBSYSTEM_SET_DISK,
+        state,
+        bucket,
+        object,
+        transaction,
+        disk_index = ?disk_index,
+        local = ?local,
+        success = ?success,
+        "rename tail probe"
+    );
+}
 pub(in crate::set_disk) const ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP: &str = "RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP";
 pub(crate) const ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE: &str = "RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE";
 /// Default reader-setup strategy for the GET read path (rustfs/backlog#1215,
@@ -3543,6 +3579,14 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
     Submit: FnOnce(rustfs_heal_contracts::heal_channel::HealChannelRequest) -> SubmitFuture + Send,
     SubmitFuture: Future<Output = ()> + Send,
 {
+    let object = request.object_prefix.as_deref().unwrap_or_default();
+    let probe = rename_probe_enabled(&request.bucket, object);
+    let probe_event = |state| {
+        if probe {
+            rename_probe(state, &request.bucket, object, "", None, None, None);
+        }
+    };
+    probe_event("tail_wait_start");
     let tail_async_started = rustfs_io_metrics::put_stage_timer();
     let (needs_heal, tail_cleanup, tail_complete) = match tail_drain.await {
         Ok(Some(outcome)) => (outcome.convergence.needs_heal(), outcome.cleanup, true),
@@ -3572,6 +3616,7 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
             (true, Vec::new(), false)
         }
     };
+    probe_event("tail_wait_end");
     rustfs_io_metrics::record_put_object_stage_duration_from(
         rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_ASYNC_WAIT,
         tail_async_started,
@@ -3580,7 +3625,10 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
     // must still run; only the explicit `false` used by the hard-crash harness
     // suppresses all in-process continuation work.
     let guard_release_wait_started = rustfs_io_metrics::put_stage_timer();
-    if matches!(guard_release.await, Ok(false)) {
+    probe_event("guard_release_wait_start");
+    let guard_release = guard_release.await;
+    probe_event("guard_release_wait_end");
+    if matches!(guard_release, Ok(false)) {
         rustfs_io_metrics::record_put_object_stage_duration_from(
             rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_GUARD_RELEASE_WAIT,
             guard_release_wait_started,
@@ -3593,7 +3641,9 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
         guard_release_wait_started,
     );
     let finalize_started = rustfs_io_metrics::put_stage_timer();
+    probe_event("finalize_start");
     finalize().await;
+    probe_event("finalize_end");
     rustfs_io_metrics::record_put_object_stage_duration_from(
         rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_FINALIZE,
         finalize_started,
@@ -3612,12 +3662,25 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
             drop(guards);
         } else if rename_tail_cleanup_defer_enabled() {
             let targets = tail_cleanup.len();
+            let probe_bucket = probe.then(|| request.bucket.clone());
+            let probe_object = probe.then(|| object.to_owned());
+            let cleanup_future = cleanup(guards, tail_cleanup);
             let job = RenameTailCleanupJob {
                 enqueued_at: Instant::now(),
                 targets,
-                future: Box::pin(cleanup(guards, tail_cleanup)),
+                future: Box::pin(async move {
+                    if let (Some(bucket), Some(object)) = (&probe_bucket, &probe_object) {
+                        rename_probe("cleanup_start", bucket, object, "", None, None, None);
+                    }
+                    cleanup_future.await;
+                    if let (Some(bucket), Some(object)) = (&probe_bucket, &probe_object) {
+                        rename_probe("cleanup_end", bucket, object, "", None, None, None);
+                    }
+                }),
             };
+            probe_event("cleanup_enqueue_start");
             if let Err(job) = rename_tail_cleanup_queue().try_enqueue(job).await {
+                probe_event("cleanup_enqueue_fallback");
                 rustfs_io_metrics::record_put_object_stage_duration_from(
                     rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_CLEANUP_SUBMIT,
                     cleanup_submit_started,
@@ -3629,6 +3692,7 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
                     cleanup_started,
                 );
             } else {
+                probe_event("cleanup_enqueued");
                 rustfs_io_metrics::record_put_object_stage_duration_from(
                     rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_CLEANUP_SUBMIT,
                     cleanup_submit_started,
@@ -3640,7 +3704,9 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
                 cleanup_submit_started,
             );
             let cleanup_started = rustfs_io_metrics::put_stage_timer();
+            probe_event("cleanup_start");
             cleanup(guards, tail_cleanup).await;
+            probe_event("cleanup_end");
             rustfs_io_metrics::record_put_object_stage_duration_from(
                 rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_CLEANUP,
                 cleanup_started,
@@ -3949,6 +4015,7 @@ impl SetDisks {
             }
         }
 
+        let probe = rename_probe_enabled(dst_bucket, dst_object);
         let disk_count = disks.len();
         let fanout_disks = disks.to_vec();
         let fanout_fence_tokens = Self::scanner_publication_lease_tokens_for_disks(disks, scanner_publication_lease_tokens)?;
@@ -3991,8 +4058,20 @@ impl SetDisks {
                             let _fanout_task_guard = Self::rename_fanout_task_guard(&dst_object);
 
                             let Some(disk) = disk else {
+                                if probe {
+                                    rename_probe(
+                                        "disk_missing",
+                                        &dst_bucket,
+                                        &dst_object,
+                                        &src_object,
+                                        Some(i),
+                                        None,
+                                        Some(false),
+                                    );
+                                }
                                 return Err(DiskError::DiskNotFound);
                             };
+                            let disk_is_local = disk.is_local();
                             let disk_wait_location_stage = if disk.is_local() {
                                 rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DISK_WAIT_LOCAL
                             } else {
@@ -4018,6 +4097,17 @@ impl SetDisks {
                             let disk_wait_started = rustfs_io_metrics::put_stage_timer();
                             dispatch_state = RenameDispatchState::MayHavePublished;
                             let _foreground_active_guard = rename_foreground_active_guard(i);
+                            if probe {
+                                rename_probe(
+                                    "disk_start",
+                                    &dst_bucket,
+                                    &dst_object,
+                                    &src_object,
+                                    Some(i),
+                                    Some(disk_is_local),
+                                    None,
+                                );
+                            }
                             let observed = disk
                                 .rename_data_borrowed_with_fence_observed(
                                     &src_bucket,
@@ -4036,6 +4126,17 @@ impl SetDisks {
                                 .await;
                             let rejected_before_publication = observed.rejected_before_publication();
                             let result = observed.result;
+                            if probe {
+                                rename_probe(
+                                    "disk_end",
+                                    &dst_bucket,
+                                    &dst_object,
+                                    &src_object,
+                                    Some(i),
+                                    Some(disk_is_local),
+                                    Some(result.is_ok()),
+                                );
+                            }
                             #[cfg(test)]
                             if result.is_ok() {
                                 rollback_fault_injection::after_rename(&dst_object, i)?;
@@ -4140,6 +4241,9 @@ impl SetDisks {
                         if let Some(commit_tx) = commit_tx.take() {
                             let _ = commit_tx.send(snapshot_commit);
                         }
+                        if probe {
+                            rename_probe("quorum", &fanout_dst_bucket, &fanout_dst_object, &fanout_src_object, None, None, None);
+                        }
                         sent_commit = true;
                         tail_after_quorum_started = rustfs_io_metrics::put_stage_timer();
                     }
@@ -4148,6 +4252,9 @@ impl SetDisks {
                     rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_AFTER_QUORUM_WAIT,
                     tail_after_quorum_started,
                 );
+                if probe {
+                    rename_probe("fanout_end", &fanout_dst_bucket, &fanout_dst_object, &fanout_src_object, None, None, None);
+                }
 
                 #[cfg(test)]
                 rollback_fault_injection::after_fanout(&fanout_dst_object);
@@ -4221,11 +4328,39 @@ impl SetDisks {
                     };
                     let dst_bucket = fanout_dst_bucket.clone();
                     let dst_object = fanout_dst_object.clone();
+                    let src_object = fanout_src_object.clone();
                     backup_reclaims.push(tokio::spawn(async move {
                         let backup_path = format!("{dst_object}/{rollback_dir}/{STORAGE_FORMAT_FILE_BACKUP}");
-                        disk.delete(&dst_bucket, &backup_path, DeleteOptions::default()).await
+                        let local = disk.is_local();
+                        if probe {
+                            rename_probe("backup_start", &dst_bucket, &dst_object, &src_object, Some(idx), Some(local), None);
+                        }
+                        let result = disk.delete(&dst_bucket, &backup_path, DeleteOptions::default()).await;
+                        if probe {
+                            rename_probe(
+                                "backup_end",
+                                &dst_bucket,
+                                &dst_object,
+                                &src_object,
+                                Some(idx),
+                                Some(local),
+                                Some(result.is_ok()),
+                            );
+                        }
+                        result
                     }));
                 }
+                if probe {
+                    rename_probe(
+                        "backups_wait_start",
+                        &fanout_dst_bucket,
+                        &fanout_dst_object,
+                        &fanout_src_object,
+                        None,
+                        None,
+                        None,
+                    );
+                }
                 for result in join_all(backup_reclaims).await {
                     match result {
                         Ok(Ok(())) => {}
@@ -4248,6 +4383,17 @@ impl SetDisks {
                         }
                     }
                 }
+                if probe {
+                    rename_probe(
+                        "backups_wait_end",
+                        &fanout_dst_bucket,
+                        &fanout_dst_object,
+                        &fanout_src_object,
+                        None,
+                        None,
+                        None,
+                    );
+                }
 
                 // The caller marks every candidate disk before returning the
                 // quorum ACK. If a refresh drains that mark while the tail is

crates/ecstore/src/set_disk/mod.rs

@@ -5129,6 +5129,12 @@ impl SetDisks {
         if request.expected_bucket_incarnation_id.is_none() {
             return Ok(());
         }
+        #[cfg(feature = "e2e-test-hooks")]
+        if let Some(object) = request.object_prefix.as_deref()
+            && skip_e2e_mrf_fallback(&request.bucket, object).await
+        {
+            return Ok(());
+        }
         request.source = rustfs_heal_contracts::heal_channel::HealRequestSource::Mrf;
         #[cfg(test)]
         {
@@ -5831,6 +5837,25 @@ impl SetDisks {
     // }
 }
 
+// The isolated root-heal test suppresses only the direct best-effort fallback.
+// Persisted MRF responsibilities and their replay keep their normal behavior.
+#[cfg(feature = "e2e-test-hooks")]
+pub(in crate::set_disk) async fn skip_e2e_mrf_fallback(bucket: &str, object: &str) -> bool {
+    let Some(path) = std::env::var_os("RUSTFS_E2E_SKIP_MRF_FALLBACK_PATH") else {
+        return false;
+    };
+    let path = std::path::PathBuf::from(path);
+    let Ok(prefix) = tokio::fs::read_to_string(&path).await else {
+        return false;
+    };
+    let key = format!("{bucket}/{object}");
+    if prefix.is_empty() || !key.starts_with(&prefix) {
+        return false;
+    }
+    let _ = tokio::fs::write(path.with_extension("observed"), key).await;
+    true
+}
+
 fn is_explicit_null_version(version_id: Option<Uuid>) -> bool {
     version_id == Some(Uuid::nil())
 }

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

@@ -9189,6 +9189,10 @@ impl SetDisks {
             // cannot safely target the object name after bucket recreation.
             return Ok(());
         };
+        #[cfg(feature = "e2e-test-hooks")]
+        if crate::set_disk::skip_e2e_mrf_fallback(bucket, object).await {
+            return Ok(());
+        }
         let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
             bucket.to_string(),
             Some(object.to_string()),

crates/lock/src/distributed_lock.rs

@@ -506,6 +506,7 @@ struct LockAcquireQuorumResult {
     individual_locks: Vec<HeldLockEntry>,
     failure_kind: Option<LockAcquireFailureKind>,
     quorum_impossible: bool,
+    attempt_timed_out: bool,
 }
 
 impl DistributedLock {
@@ -886,9 +887,10 @@ impl DistributedLock {
     }
 
     async fn acquire_lock_quorum_with_retry(&self, request: &LockRequest) -> Result<LockAcquireQuorumResult> {
-        let start = std::time::Instant::now();
+        let start = tokio::time::Instant::now();
         let mut attempt = 1;
-        let mut last_result = None;
+        let mut last_result: Option<LockAcquireQuorumResult> = None;
+        let mut preserve_quorum_on_attempt_timeout = false;
 
         loop {
             let elapsed = start.elapsed();
@@ -906,13 +908,23 @@ impl DistributedLock {
                 return Ok(result);
             }
 
-            last_result = Some(result);
-            let backoff = Self::lock_acquire_retry_backoff(attempt, &mut rand::rng());
-            if start.elapsed().saturating_add(backoff) >= request.acquire_timeout {
+            // A shortened final retry may expire without establishing whether
+            // the previously observed quorum failure has recovered.
+            if !preserve_quorum_on_attempt_timeout || !result.attempt_timed_out {
+                last_result = Some(result);
+                preserve_quorum_on_attempt_timeout = false;
+            }
+            let remaining = request.acquire_timeout.saturating_sub(start.elapsed());
+            if remaining.is_zero() {
                 break;
             }
-
-            tokio::time::sleep(backoff).await;
+            let backoff = Self::lock_acquire_retry_backoff(attempt, &mut rand::rng());
+            if backoff >= remaining && last_result.as_ref().is_some_and(|result| result.quorum_impossible) {
+                preserve_quorum_on_attempt_timeout = true;
+            }
+            // Keep time for another acquire attempt instead of abandoning the
+            // caller's remaining budget when the jittered backoff grows large.
+            tokio::time::sleep(backoff.min(remaining / 2)).await;
             attempt += 1;
         }
 
@@ -927,6 +939,7 @@ impl DistributedLock {
                 individual_locks: Vec::new(),
                 failure_kind: Some(LockAcquireFailureKind::RetryableContention),
                 quorum_impossible: false,
+                attempt_timed_out: true,
             })
         }
     }
@@ -937,6 +950,7 @@ impl DistributedLock {
             individual_locks: Vec::new(),
             failure_kind: Some(LockAcquireFailureKind::RetryableContention),
             quorum_impossible: false,
+            attempt_timed_out: true,
         }
     }
 
@@ -966,6 +980,7 @@ impl DistributedLock {
             individual_locks,
             failure_kind: Some(failure_kind),
             quorum_impossible: matches!(failure_kind, LockAcquireFailureKind::UnrecoverableQuorum),
+            attempt_timed_out: true,
         }
     }
 
@@ -1134,6 +1149,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: Some(failure_kind),
                     quorum_impossible: true,
+                    attempt_timed_out: false,
                 });
             }
 
@@ -1176,6 +1192,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: None,
                     quorum_impossible: false,
+                    attempt_timed_out: false,
                 });
             }
 
@@ -1221,6 +1238,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: Some(failure_kind),
                     quorum_impossible,
+                    attempt_timed_out: false,
                 });
             }
         }
@@ -1245,6 +1263,7 @@ impl DistributedLock {
                 last_failure_kind.unwrap_or(LockAcquireFailureKind::RetryableContention)
             }),
             quorum_impossible: hard_failures > 0,
+            attempt_timed_out: false,
         })
     }
 }
@@ -2246,6 +2265,71 @@ mod tests {
         );
     }
 
+    fn clients_with_initial_quorum_loss(retries: &[AcquirePlan]) -> Vec<Arc<dyn LockClient>> {
+        (0..4)
+            .map(|index| {
+                let first = if index < 2 {
+                    AcquirePlan::Success { delay: Duration::ZERO }
+                } else {
+                    AcquirePlan::Failure {
+                        error: "Remote lock RPC failed: node unavailable",
+                        delay: Duration::ZERO,
+                    }
+                };
+                let mut plans = vec![first];
+                plans.extend_from_slice(retries);
+                Arc::new(SequencedClient::new(plans, Arc::new(Mutex::new(Vec::new())))) as Arc<dyn LockClient>
+            })
+            .collect()
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acquire_guard_preserves_quorum_failure_when_last_retry_times_out() {
+        let clients = clients_with_initial_quorum_loss(&[AcquirePlan::Success {
+            delay: Duration::from_secs(1),
+        }]);
+        let lock = DistributedLock::new("test".to_string(), clients, 3);
+        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
+            .with_acquire_timeout(Duration::from_millis(180));
+
+        let result = lock.acquire_guard(&request).await;
+        assert!(
+            matches!(result, Err(LockError::QuorumNotReached { required: 3, .. })),
+            "a late inconclusive retry must not hide an observed quorum failure: {result:?}"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acquire_guard_recovers_quorum_with_shortened_retry() {
+        let clients = clients_with_initial_quorum_loss(&[AcquirePlan::Success { delay: Duration::ZERO }]);
+        let lock = DistributedLock::new("test".to_string(), clients, 3);
+        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
+            .with_acquire_timeout(Duration::from_millis(180));
+
+        let guard = lock
+            .acquire_guard(&request)
+            .await
+            .expect("a retry should recover from quorum loss")
+            .expect("a recovered quorum should acquire the lock");
+        assert_eq!(guard.entries.len(), 3);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acquire_guard_replaces_quorum_failure_with_later_contention() {
+        let clients = clients_with_initial_quorum_loss(
+            &[AcquirePlan::Failure {
+                error: "Lock acquisition timeout",
+                delay: Duration::ZERO,
+            }; 64],
+        );
+        let lock = DistributedLock::new("test".to_string(), clients, 3);
+        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
+            .with_acquire_timeout(Duration::from_millis(180));
+
+        let result = lock.acquire_guard(&request).await;
+        assert!(matches!(result, Ok(None)), "later contention should replace the earlier quorum failure");
+    }
+
     #[tokio::test]
     async fn acquire_guard_retries_rpc_timeouts_until_caller_timeout() {
         let clients: Vec<Arc<dyn LockClient>> = vec![

crates/lock/src/namespace/tests.rs

@@ -979,6 +979,30 @@ async fn test_namespace_lock_distributed_retries_transient_acquire_timeout() {
     drop(guard);
 }
 
+#[tokio::test(start_paused = true)]
+async fn test_namespace_lock_retries_when_backoff_exceeds_remaining_budget() {
+    let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let flaky_clients = managers
+        .iter()
+        .map(|manager| Arc::new(FlakyAcquireClient::new(manager.clone(), 1)))
+        .collect::<Vec<_>>();
+    let clients = flaky_clients
+        .iter()
+        .map(|client| client.clone() as Arc<dyn LockClient>)
+        .collect::<Vec<_>>();
+    let lock = NamespaceLock::with_clients("short-retry-budget".to_string(), clients);
+    let resource = create_test_object_key("bucket", "short-retry-budget");
+
+    // The first jittered backoff is at least 187 ms, but a second attempt can
+    // still succeed within this budget once the transient failure has cleared.
+    let guard = lock
+        .get_write_lock(resource, "owner", Duration::from_millis(180))
+        .await
+        .expect("a backoff longer than the remaining budget must not discard the remaining attempt");
+    assert!(flaky_clients.iter().any(|client| client.acquire_attempts() >= 2));
+    drop(guard);
+}
+
 #[tokio::test]
 async fn test_namespace_lock_distributed_waits_full_timeout_for_late_release() {
     let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
@@ -1136,10 +1160,16 @@ async fn test_namespace_lock_distributed_contention_quorum_miss_times_out() {
     let lock = NamespaceLock::with_clients("contention-timeout".to_string(), clients);
     let resource = create_test_object_key("bucket", "object-contention-timeout");
 
-    let err = lock
-        .get_write_lock(resource, "owner-a", Duration::from_millis(40))
+    let timeout = Duration::from_millis(40);
+    let started = std::time::Instant::now();
+    let err = tokio::time::timeout(Duration::from_secs(2), lock.get_write_lock(resource, "owner-a", timeout))
         .await
+        .expect("contended acquisition must remain bounded")
         .expect_err("ordinary contention should exhaust acquire timeout");
+    assert!(
+        started.elapsed() >= timeout,
+        "a contended acquisition must not return before its configured deadline"
+    );
 
     let err_str = err.to_string().to_lowercase();
     assert!(err_str.contains("timeout"), "expected timeout after contention retries, got: {err}");