fix(notification-outbox): keyset Pending back-fill + distinct ids so a mid-claim Retry cannot double-deliver (review 2026-10-21 06/PERF-5 follow-up)
This commit is contained in:
+20
-2
@@ -172,14 +172,32 @@ VALUES
|
||||
if (remaining > 0 && pending.Count == batchSize - retryingQuota)
|
||||
{
|
||||
// Retrying could not fill its share; back-fill with newer Pending rows.
|
||||
// KEYSET, not OFFSET: the claim spans several queries with no lock between
|
||||
// them, so a row can turn Pending in the gap — an operator Retry flips an
|
||||
// older Parked row back to Pending with its original CreatedAt. An OFFSET
|
||||
// back-fill (Skip(pending.Count)) would then shift by one and re-read the
|
||||
// last row already claimed, and the sweep would deliver that notification
|
||||
// twice in parallel. Resuming strictly after the last claimed
|
||||
// (CreatedAt, NotificationId) — the ORDER BY's own key — cannot overlap.
|
||||
var last = pending[^1];
|
||||
var lastCreatedAt = last.CreatedAt;
|
||||
var lastId = last.NotificationId;
|
||||
pending.AddRange(await DuePendingQuery()
|
||||
.Skip(pending.Count)
|
||||
.Where(n => n.CreatedAt > lastCreatedAt
|
||||
|| (n.CreatedAt == lastCreatedAt && string.Compare(n.NotificationId, lastId) > 0))
|
||||
.Take(remaining)
|
||||
.ToListAsync(cancellationToken));
|
||||
}
|
||||
|
||||
pending.AddRange(retrying);
|
||||
return pending.OrderBy(n => n.CreatedAt).ToList();
|
||||
|
||||
// Belt and braces: a row can also change status between the Pending and
|
||||
// Retrying reads, so never hand the dispatcher the same id twice.
|
||||
return pending
|
||||
.DistinctBy(n => n.NotificationId)
|
||||
.OrderBy(n => n.CreatedAt)
|
||||
.ThenBy(n => n.NotificationId, StringComparer.Ordinal)
|
||||
.ToList();
|
||||
|
||||
IQueryable<Notification> DuePendingQuery() => _context.Notifications
|
||||
.AsNoTracking()
|
||||
|
||||
@@ -632,6 +632,80 @@ public class NotificationOutboxRepositoryTests : IDisposable
|
||||
due.Select(n => n.NotificationId).ToArray());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that a row turning Pending BETWEEN the claim's queries (an operator Retry
|
||||
/// flipping an older Parked row back to Pending) cannot shift the back-fill onto a row
|
||||
/// already claimed: the batch never contains a duplicate id, so no notification is
|
||||
/// delivered twice in one sweep (adversarial review of 06/PERF-5).
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task GetDueAsync_RowTurnsPendingMidClaim_NeverClaimsAnIdTwice()
|
||||
{
|
||||
var now = new DateTimeOffset(2026, 5, 19, 12, 0, 0, TimeSpan.Zero);
|
||||
var retried = MakeNotification(Guid.NewGuid().ToString(), NotificationStatus.Parked,
|
||||
createdAt: now.AddHours(-5));
|
||||
var pending = Enumerable.Range(0, 5)
|
||||
.Select(i => MakeNotification(Guid.NewGuid().ToString(), NotificationStatus.Pending,
|
||||
createdAt: now.AddMinutes(-10 + i)))
|
||||
.ToList();
|
||||
|
||||
var flip = new FlipToPendingBeforeSecondRead(retried.NotificationId);
|
||||
await using var ctx = SqliteTestHelper.CreateInMemoryContext(flip);
|
||||
ctx.Notifications.Add(retried);
|
||||
ctx.Notifications.AddRange(pending);
|
||||
await ctx.SaveChangesAsync();
|
||||
ctx.ChangeTracker.Clear();
|
||||
flip.Arm();
|
||||
|
||||
var due = await new NotificationOutboxRepository(ctx).GetDueAsync(now, batchSize: 4);
|
||||
|
||||
var ids = due.Select(n => n.NotificationId).ToList();
|
||||
Assert.Equal(ids.Count, ids.Distinct().Count());
|
||||
Assert.Equal(
|
||||
new[]
|
||||
{
|
||||
pending[0].NotificationId, pending[1].NotificationId,
|
||||
pending[2].NotificationId, pending[3].NotificationId,
|
||||
},
|
||||
ids);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Flips one notification to Pending just before the claim's SECOND read against the
|
||||
/// Notifications table — the window an operator Retry can land in.
|
||||
/// </summary>
|
||||
private sealed class FlipToPendingBeforeSecondRead(string notificationId)
|
||||
: Microsoft.EntityFrameworkCore.Diagnostics.DbCommandInterceptor
|
||||
{
|
||||
private int _reads = -1;
|
||||
|
||||
/// <summary>Starts counting reads from the next command on.</summary>
|
||||
public void Arm() => _reads = 0;
|
||||
|
||||
/// <inheritdoc />
|
||||
public override async ValueTask<Microsoft.EntityFrameworkCore.Diagnostics.InterceptionResult<System.Data.Common.DbDataReader>> ReaderExecutingAsync(
|
||||
System.Data.Common.DbCommand command,
|
||||
Microsoft.EntityFrameworkCore.Diagnostics.CommandEventData eventData,
|
||||
Microsoft.EntityFrameworkCore.Diagnostics.InterceptionResult<System.Data.Common.DbDataReader> result,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_reads >= 0 && command.CommandText.Contains("Notifications", StringComparison.Ordinal)
|
||||
&& ++_reads == 2)
|
||||
{
|
||||
await using var flip = command.Connection!.CreateCommand();
|
||||
flip.CommandText = "UPDATE Notifications SET Status = 'Pending' WHERE NotificationId = $id;";
|
||||
var p = flip.CreateParameter();
|
||||
p.ParameterName = "$id";
|
||||
p.Value = notificationId;
|
||||
flip.Parameters.Add(p);
|
||||
await flip.ExecuteNonQueryAsync(cancellationToken);
|
||||
}
|
||||
|
||||
return await base.ReaderExecutingAsync(command, eventData, result, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Verifies that get by id returns the row when present and null otherwise.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
|
||||
Reference in New Issue
Block a user