Persist deduplicated message positions so they are always cleared

[stidsborg]

Summary

When QueueManager.ProcessMessages drops a message as a duplicate (by idempotency key), it now stages the position into _deliveredPositions and persists it to the DeliveredPositionsId effect immediately, instead of relying on an immediate per-position store delete.

Why

The previous edit changed the dedup branch from await _messageClearer.Clear([position]) to _deliveredPositions.Add(position). The problem: _deliveredPositions is only written into the DeliveredPositionsId effect inside DeliverMessages (on a real delivery), and both AfterFlush and Initialize clear from the effect, not the in-memory set.

So if no delivery followed in the same QueueManager lifetime — an all-duplicate batch, the non-dup message never matching a subscription, or the flow suspending/completing first — the duplicate's position:

• was never written to the DeliveredPositionsId effect,
• so AfterFlush never deleted it from the store, and
• a crash could not replay it (Initialize reads the effect, which never had it).

The duplicate could linger in the message store and be repeatedly re-fetched across replica restarts.

Fix

Persist the position into the effect at stage time, routing dedup cleanup through the same batched, crash-safe drain path (effect → AfterFlush → MessageClearer.Clear) as real deliveries. The write is taken under _lock to match the DeliverMessages access to _deliveredPositions.

if (idempotencyKey != null && !_idempotencyKeys.Add(idempotencyKey, position))
{
    lock (_lock)
    {
        _deliveredPositions.Add(position);
        _effect.FlushlessUpsert(DeliveredPositionsId, _deliveredPositions.ToList(), alias: null);
    }
    continue;
}

Testing

• dotnet build of the core project passes (0 warnings, 0 errors).