From 5f7f889a9678d4c3a1906648e6d5a24f7b5b4614 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 09:49:19 -0400 Subject: [PATCH] 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) --- .../NotificationOutboxRepository.cs | 22 +++++- .../RepositoryCoverageTests.cs | 74 +++++++++++++++++++ 2 files changed, 94 insertions(+), 2 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs index 6601e9c7..0bed52e2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs @@ -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 DuePendingQuery() => _context.Notifications .AsNoTracking() diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs index e0dd8b40..06728611 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs @@ -632,6 +632,80 @@ public class NotificationOutboxRepositoryTests : IDisposable due.Select(n => n.NotificationId).ToArray()); } + /// + /// 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). + /// + /// A task that represents the asynchronous operation. + [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); + } + + /// + /// Flips one notification to Pending just before the claim's SECOND read against the + /// Notifications table — the window an operator Retry can land in. + /// + private sealed class FlipToPendingBeforeSecondRead(string notificationId) + : Microsoft.EntityFrameworkCore.Diagnostics.DbCommandInterceptor + { + private int _reads = -1; + + /// Starts counting reads from the next command on. + public void Arm() => _reads = 0; + + /// + public override async ValueTask> ReaderExecutingAsync( + System.Data.Common.DbCommand command, + Microsoft.EntityFrameworkCore.Diagnostics.CommandEventData eventData, + Microsoft.EntityFrameworkCore.Diagnostics.InterceptionResult 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); + } + } + /// Verifies that get by id returns the row when present and null otherwise. /// A task that represents the asynchronous operation. [Fact]