test(heal): wait for metadata write tail

.config/nextest.toml

@@ -960,6 +960,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) {
@@ -1952,6 +1961,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
@@ -2045,7 +2069,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 {
@@ -2542,6 +2571,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/mod.rs

@@ -5154,6 +5154,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)]
         {
@@ -5861,6 +5867,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/heal.rs

@@ -1943,6 +1943,13 @@ impl SetDisks {
                             }
                         }
                         let _read_repair_write_lock_guard = if let Some(expected_fingerprint) = read_repair_commit_fingerprint {
+                            if _read_repair_read_lock_guard
+                                .as_ref()
+                                .is_some_and(|guard| guard.is_lock_lost())
+                            {
+                                let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
+                                return Err(DiskError::DiskOngoingReq);
+                            }
                             drop(_read_repair_read_lock_guard.take());
                             match self
                                 .acquire_revalidated_read_repair_commit_lock(
@@ -1985,7 +1992,12 @@ impl SetDisks {
                                 if let Some(scope) = &bucket_heal_scope {
                                     scope.check()?;
                                 }
-                                if commit_fence.is_some_and(|check| !check()) {
+                                if commit_fence.is_some_and(|check| !check())
+                                    || _write_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
+                                    || _read_repair_write_lock_guard
+                                        .as_ref()
+                                        .is_some_and(|guard| guard.is_lock_lost())
+                                {
                                     let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
                                     return Err(DiskError::DiskOngoingReq);
                                 }
@@ -2123,6 +2135,12 @@ impl SetDisks {
                         self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id)
                             .await
                             .map_err(DiskError::other)?;
+                        if _read_repair_write_lock_guard
+                            .as_ref()
+                            .is_some_and(|guard| guard.is_lock_lost())
+                        {
+                            return Err(DiskError::DiskOngoingReq);
+                        }
 
                         self.record_healed_capacity_scope(&healed_disks);
 
@@ -3625,6 +3643,46 @@ impl SetDisks {
         proof: AbsenceProofRequest<'_>,
     ) -> Result<(HealResultItem, Option<Error>)> {
         *absence = None;
+        if opts.read_repair
+            && !opts.no_lock
+            && !opts.dry_run
+            && !opts.remove
+            && !has_suffix(object, SLASH_SEPARATOR)
+            && !crate::bucket::utils::is_meta_bucketname(bucket)
+        {
+            // Read repair may stage under a shared lock, but must not perform
+            // destructive cleanup or produce an authoritative absence proof.
+            let can_commit = || proof.retirement.is_none_or(|context| !context.lifecycle_guard.is_lock_lost());
+            let mut read_opts = *opts;
+            loop {
+                let (result, error) = pin_object_heal_future(|| {
+                    self.heal_object_with_explicit_version_regen(
+                        bucket,
+                        object,
+                        version_id,
+                        ExplicitVersionHeal {
+                            opts: &read_opts,
+                            allow_regeneration: false,
+                            preserve_unrepairable: true,
+                            commit_fence: Some(&can_commit),
+                            namespace_owner: None,
+                            retirement: None,
+                        },
+                        absence,
+                    )
+                })
+                .await
+                .map_err(|error| to_object_err(error.into(), vec![bucket, object]))?;
+                if !can_commit() {
+                    return Err(DiskError::DiskOngoingReq.into());
+                }
+                if matches!(error, Some(DiskError::FileCorrupt)) && read_opts.scan_mode != HealScanMode::Deep {
+                    read_opts.scan_mode = HealScanMode::Deep;
+                    continue;
+                }
+                return Ok((result, error.map(Into::into)));
+            }
+        }
         let _write_lock_guard = if !opts.no_lock {
             let ns_lock = self
                 .new_ns_lock(bucket, object)
@@ -8185,7 +8243,7 @@ mod heal_result_report_tests {
             let heal_bucket = bucket.clone();
             let heal = tokio::spawn(async move {
                 heal_set
-                    .heal_object(
+                    .heal_object_with_retirement(
                         &heal_bucket,
                         object,
                         &selected_version.to_string(),
@@ -8194,6 +8252,11 @@ mod heal_result_report_tests {
                             read_repair: true,
                             ..Default::default()
                         },
+                        &mut None,
+                        AbsenceProofRequest {
+                            retirement: None,
+                            allow_unversioned: false,
+                        },
                     )
                     .await
             });
@@ -8285,7 +8348,7 @@ mod heal_result_report_tests {
                 assert!(outcome.expect("historical version remains healable").1.is_none());
             } else {
                 assert!(
-                    matches!(outcome, Err(DiskError::FileVersionNotFound)),
+                    matches!(outcome, Err(Error::VersionNotFound(..))),
                     "deleted version must remain typed absent: {outcome:?}"
                 );
             }
@@ -9877,6 +9940,123 @@ mod heal_result_report_tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn read_repair_storage_entry_keeps_readers_available_during_reconstruction() {
+        use crate::object_api::WriteCompletion;
+        use std::time::Duration;
+        use tokio::io::AsyncReadExt;
+
+        let (dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "read-repair-storage-entry";
+        let object = "object.bin";
+        let expected = vec![0x37; 1024 * 1024];
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        set.put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(expected.clone()),
+            &ObjectOptions {
+                write_completion: WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("write object");
+        let metadata = disks[0]
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("read metadata");
+        let missing = dirs[0]
+            .path()
+            .join(bucket)
+            .join(object)
+            .join(metadata.data_dir.expect("external object data").to_string())
+            .join("part.1");
+        tokio::fs::remove_file(&missing).await.expect("remove shard");
+
+        let (_pause, arrived, release) =
+            ReadRepairPauseScope::install_at(bucket, object, super::ReadRepairPausePhase::PartStaging(1));
+        let heal_set = set.clone();
+        let heal = tokio::spawn(async move {
+            let mut absence = None;
+            let result = heal_set
+                .heal_object_with_retirement(
+                    bucket,
+                    object,
+                    "",
+                    &HealOpts {
+                        scan_mode: HealScanMode::Deep,
+                        read_repair: true,
+                        ..Default::default()
+                    },
+                    &mut absence,
+                    AbsenceProofRequest {
+                        retirement: None,
+                        allow_unversioned: false,
+                    },
+                )
+                .await;
+            (result, absence)
+        });
+        tokio::time::timeout(Duration::from_secs(10), arrived.notified())
+            .await
+            .expect("read-repair reached shard reconstruction");
+        let namespace = set.new_ns_lock(bucket, object).await.expect("object namespace");
+        let reader = namespace
+            .get_read_lock(Duration::from_millis(200))
+            .await
+            .expect("read-repair reconstruction should allow concurrent readers");
+        drop(reader);
+        release.notify_one();
+        let (result, absence) = tokio::time::timeout(Duration::from_secs(20), heal)
+            .await
+            .expect("repair finishes")
+            .expect("repair task joins");
+        let (_, error) = result.expect("repair operation succeeds");
+        assert!(error.is_none(), "repair failed: {error:?}");
+        assert!(absence.is_none());
+        assert!(missing.exists(), "missing shard was not restored");
+        let mut reader = set
+            .get_object_reader(bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("read repaired object");
+        let mut actual = Vec::new();
+        reader.stream.read_to_end(&mut actual).await.expect("read full object");
+        assert_eq!(actual, expected);
+    }
+
+    #[tokio::test]
+    async fn read_repair_storage_entry_does_not_prove_missing_versions_absent() {
+        let (_dirs, _disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "read-repair-no-absence-proof";
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        let mut absence = None;
+        let (_, error) = set
+            .heal_object_with_retirement(
+                bucket,
+                "missing.bin",
+                &Uuid::new_v4().to_string(),
+                &HealOpts {
+                    read_repair: true,
+                    ..Default::default()
+                },
+                &mut absence,
+                AbsenceProofRequest {
+                    retirement: None,
+                    allow_unversioned: false,
+                },
+            )
+            .await
+            .expect("missing version report");
+        assert!(matches!(error, Some(Error::FileVersionNotFound)));
+        assert!(absence.is_none(), "read repair must not publish an absence proof");
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn read_repair_commit_aborts_when_object_changes_before_final_rename() {
@@ -9922,7 +10102,7 @@ mod heal_result_report_tests {
         let heal_set = set.clone();
         let heal_task = tokio::spawn(async move {
             heal_set
-                .heal_object(
+                .heal_object_with_retirement(
                     bucket,
                     object,
                     "",
@@ -9931,6 +10111,11 @@ mod heal_result_report_tests {
                         read_repair: true,
                         ..Default::default()
                     },
+                    &mut None,
+                    AbsenceProofRequest {
+                        retirement: None,
+                        allow_unversioned: false,
+                    },
                 )
                 .await
         });
@@ -9951,7 +10136,7 @@ mod heal_result_report_tests {
             .expect("read-repair should finish after release")
             .expect("read-repair task should join")
             .expect("stale read-repair should report a typed retryable error");
-        assert_eq!(err, Some(DiskError::Timeout));
+        assert!(matches!(err, Some(Error::Timeout)), "stale repair must retry: {err:?}");
 
         let final_info = set
             .get_object_info(bucket, object, &ObjectOptions::default())

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

@@ -9262,6 +9262,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/ecstore/src/store/heal.rs

@@ -2845,9 +2845,18 @@ mod tests {
         let (configuration, _) = read_config_no_lock_preserve_empty_with_metadata(store.clone(), &object)
             .await
             .expect("read the current bucket configuration");
-        save_config(store.clone(), &object, configuration)
-            .await
-            .expect("publish a newer config revision in the same bucket incarnation");
+        crate::config::com::save_config_with_opts(
+            store.clone(),
+            &object,
+            configuration,
+            &ObjectOptions {
+                max_parity: true,
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("publish a newer config revision on the target disk");
         let committed =
             std::fs::read(target_path.join(RUSTFS_META_BUCKET).join(&object).join("xl.meta")).expect("new target configuration");
         assert!(

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,9 @@ 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;
 
         loop {
             let elapsed = start.elapsed();
@@ -906,13 +907,22 @@ 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 final retry can time out without establishing whether the
+            // previously observed quorum failure has recovered.
+            let remaining = request.acquire_timeout.saturating_sub(start.elapsed());
+            let preserve_quorum_failure = result.attempt_timed_out
+                && remaining.is_zero()
+                && last_result.as_ref().is_some_and(|result| result.quorum_impossible);
+            if !preserve_quorum_failure {
+                last_result = Some(result);
+            }
+            if remaining.is_zero() {
                 break;
             }
-
-            tokio::time::sleep(backoff).await;
+            let backoff = Self::lock_acquire_retry_backoff(attempt, &mut rand::rng());
+            // 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 +937,7 @@ impl DistributedLock {
                 individual_locks: Vec::new(),
                 failure_kind: Some(LockAcquireFailureKind::RetryableContention),
                 quorum_impossible: false,
+                attempt_timed_out: true,
             })
         }
     }
@@ -937,6 +948,7 @@ impl DistributedLock {
             individual_locks: Vec::new(),
             failure_kind: Some(LockAcquireFailureKind::RetryableContention),
             quorum_impossible: false,
+            attempt_timed_out: true,
         }
     }
 
@@ -966,6 +978,7 @@ impl DistributedLock {
             individual_locks,
             failure_kind: Some(failure_kind),
             quorum_impossible: matches!(failure_kind, LockAcquireFailureKind::UnrecoverableQuorum),
+            attempt_timed_out: true,
         }
     }
 
@@ -1134,6 +1147,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: Some(failure_kind),
                     quorum_impossible: true,
+                    attempt_timed_out: false,
                 });
             }
 
@@ -1176,6 +1190,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: None,
                     quorum_impossible: false,
+                    attempt_timed_out: false,
                 });
             }
 
@@ -1221,6 +1236,7 @@ impl DistributedLock {
                     individual_locks,
                     failure_kind: Some(failure_kind),
                     quorum_impossible,
+                    attempt_timed_out: false,
                 });
             }
         }
@@ -1245,6 +1261,7 @@ impl DistributedLock {
                 last_failure_kind.unwrap_or(LockAcquireFailureKind::RetryableContention)
             }),
             quorum_impossible: hard_failures > 0,
+            attempt_timed_out: false,
         })
     }
 }
@@ -2246,6 +2263,87 @@ 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_preserves_quorum_failure_when_retry_backoff_fits_but_attempt_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(500));
+
+        let result = lock.acquire_guard(&request).await;
+        assert!(
+            matches!(result, Err(LockError::QuorumNotReached { required: 3, .. })),
+            "an inconclusive final attempt 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}");