diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs
index 202d4ed18..517d13117 100644
--- a/crates/ecstore/src/set_disk/ops/object.rs
+++ b/crates/ecstore/src/set_disk/ops/object.rs
@@ -10020,6 +10020,17 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 && !guarded
                 && goi.name.is_empty()
             {
+                // Recovery can restore or retire versions of this key
+                // (rolling back an interrupted prepared delete brings a
+                // retained version back). A pre-fetched read for a later
+                // duplicate of the same name would still report the stale
+                // miss and skip its object-lock check — bypassing retention —
+                // so evict those entries and let the loop re-read them live.
+                for (other_index, other) in objects.iter().enumerate() {
+                    if other_index != i && other.object_name == dobj.object_name {
+                        source_reads.remove(&other_index);
+                    }
+                }
                 let recovery = self
                     .recover_delete_key(
                         bucket,
diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs
index 0f271722b..e9a05ef6b 100644
--- a/crates/lock/src/distributed_lock.rs
+++ b/crates/lock/src/distributed_lock.rs
@@ -810,10 +810,12 @@ impl DistributedLock {
             }
         }
 
-        // Awaited so a caller that falls back per key cannot race these
-        // unlocks (the distributed release on drop is fire-and-forget).
+        // One bounded synchronous pass so the caller cannot race these
+        // unlocks on an immediate retry; anything a slow peer still holds
+        // keeps retrying in the background instead of parking the request
+        // on the deferred unlock schedule.
         if !orphaned_entries.is_empty() {
-            Self::release_entries(orphaned_entries, "distributed_lock_batch_quorum_rollback").await;
+            Self::release_entries_bounded(orphaned_entries, "distributed_lock_batch_quorum_rollback").await;
         }
         results
     }
@@ -879,6 +881,19 @@ impl DistributedLock {
         Self::release_entries_deferred(pending, context).await;
     }
 
+    /// Release entries with one synchronous pass, then continue the retry
+    /// schedule in the background.
+    ///
+    /// The batch-acquire rollback path runs on the request: it must not park
+    /// the caller on the deferred unlock schedule when a peer is unreachable,
+    /// while still confirming the healthy quorum is free before returning.
+    async fn release_entries_bounded(entries: Vec<(LockId, Arc<dyn LockClient>)>, context: &'static str) {
+        let pending = Self::release_pending_once(entries, 1, context).await;
+        if !pending.is_empty() {
+            Self::spawn_release_cleanup(pending, context);
+        }
+    }
+
     async fn release_pending_once(
         pending: Vec<(LockId, Arc<dyn LockClient>)>,
         attempt: usize,