From b8bd0e9f15aaa236367128685d7f29ffaa409141 Mon Sep 17 00:00:00 2001 From: Hauser Date: Wed, 4 Nov 2026 16:55:00 +0800 Subject: [PATCH] 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 | 11 +++++++++++ crates/lock/src/distributed_lock.rs | 21 ++++++++++++++++++--- 2 files changed, 29 insertions(+), 3 deletions(-) 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)>, 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)>, attempt: usize,