fix(ecstore,lock): refresh duplicate keys after recovery, bound unlock

Review findings on the batched lock rounds. A batch entry whose recovery
rolls back an interrupted prepared delete can bring a retained version
back; a later duplicate of the same key still held its pre-fetched
source-missing read, skipped the object-lock check, and deleted the
restored version. Recovery now evicts the pre-fetched reads of every
other entry naming the same object, so those entries take the loop's
live re-read and see the restored version.

The batch rollback release awaited the full unlock retry schedule,
including the deferred background pass, parking the request behind an
unreachable peer. The synchronous phase is now a single release pass
with the deferred retries spawned in the background.

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,

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,