Skip to content

Commit 4568ce6

Browse files
committed
Prevent duplicate delivery of a re-fetched already-delivered message
Under push-only delivery a delivered message position can be re-fetched (a reopen removes it from the clearer's ignore-set before the asynchronous Clear deletes it from the store) and routed to a fresh QueueManager on restart. That instance reloaded the idempotency keys but discarded the delivered-positions, so IdempotencyKeys.Add(sameKey, samePosition) returned true and the message was delivered a second time. Seed _fetchedPositions from the persisted DeliveredPositionsId on Initialize so the existing ProcessMessages skip-check catches a re-fetched delivered position. Only delivered/skipped positions are seeded, so a recorded-but-not-delivered message still re-delivers correctly - no message loss.
1 parent 06f0bf4 commit 4568ce6

1 file changed

Lines changed: 6 additions & 0 deletions

File tree

Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,12 @@ private async Task Initialize()
8686

8787
if (_effect.TryGet<List<long>>(DeliveredPositionsId, out var positions) && positions is { Count: > 0 })
8888
{
89+
// Remember the positions a previous instance already delivered, so a message that is re-fetched
90+
// before its Clear deletes it from the store is skipped here rather than delivered a second time.
91+
lock (_lock)
92+
foreach (var position in positions)
93+
_fetchedPositions.Add(position);
94+
8995
await _messageClearer.Clear(positions);
9096
positions.Clear();
9197
_effect.FlushlessUpsert(DeliveredPositionsId, positions, alias: null);

0 commit comments

Comments
 (0)