fix(notifications): list sort index + CAS operator Retry/Discard (P5-3, P5-4)
P5-3: add IX_Notifications_Created_NotificationId (CreatedAt DESC, NotificationId DESC) — the Central UI Notification Outbox list page sorts on exactly that pair, but no existing index leads unconditionally on CreatedAt, so an unfiltered/lightly filtered page fell back to a full clustered-index scan + sort on every COUNT + page request. Mirrors the Medium wave's IX_SiteCalls_Created_TrackedOperationId migration + SQL twin pattern faithfully, including the handwritten docs/plans/sql script. P5-4: operator Retry/Discard was read-check-write keyed on NotificationId only — a detached PipeTo task interleaving with the dispatcher's own write (or a second concurrent operator action) could land between the read and the write and get silently clobbered by the unconditional UpdateAsync, stamping a Delivered row Discarded with stale fields. Adds INotificationOutboxRepository.TryUpdateFromParkedAsync, a single-statement compare-and-swap (WHERE Status == Parked) mirroring SecuredWriteRepository.TryMarkApprovedAsync's CAS shape. NotificationOutboxActor's RetryAsync/DiscardAsync now use the CAS write and, on a lost race, re-read to report "notification already progressed to <Status>" instead of silently overwriting whatever is there now. Proof of failure: reverting only the actor's CAS usage while keeping the repository addition reproduces the silent-overwrite bug — 7 tests fail, including two new deterministic (non-sleep) race-proof tests that mock the exact GetByIdAsync -> lost-CAS -> re-read sequence the finding describes. P5-3 is perf-only; verified via migration/SQL-twin parity instead (no runtime proof-of-failure is meaningful for an index).
This commit is contained in:
@@ -0,0 +1,51 @@
|
|||||||
|
-- arch-review 2026-08-19 P5-3 (register row-35 class): the Notifications list
|
||||||
|
-- page's sort key gets its own index.
|
||||||
|
--
|
||||||
|
-- NotificationOutboxRepository.QueryAsync orders (CreatedAt DESC, NotificationId
|
||||||
|
-- DESC) (NotificationId breaks CreatedAt ties for deterministic OFFSET paging).
|
||||||
|
-- None of the existing Notifications indexes lead unconditionally on CreatedAt:
|
||||||
|
-- - IX_Notifications_Status_NextAttemptAt requires an equality on Status first.
|
||||||
|
-- - IX_Notifications_SourceSiteId_CreatedAt requires an equality on SourceSiteId first.
|
||||||
|
-- - IX_Notifications_Delivered is filtered to Status = 'Delivered' only.
|
||||||
|
-- So an unfiltered (or lightly filtered) page fell back to a full clustered-index
|
||||||
|
-- scan + sort. IX_Notifications_Created_NotificationId (CreatedAt DESC,
|
||||||
|
-- NotificationId DESC) makes a forward scan of the index deliver rows in the
|
||||||
|
-- requested order directly, so a page reads a Top-N off the index instead of
|
||||||
|
-- sorting the whole table on every COUNT + page request.
|
||||||
|
--
|
||||||
|
-- Notifications is small enough today that a plain (offline) index build is
|
||||||
|
-- fine. On an Enterprise edition instance where the table has grown large and
|
||||||
|
-- the Central UI must stay responsive, add WITH (ONLINE = ON) to the CREATE
|
||||||
|
-- INDEX below.
|
||||||
|
--
|
||||||
|
-- Run with QUOTED_IDENTIFIER ON. Several tables in this database carry filtered
|
||||||
|
-- indexes and persisted computed columns, and SQL Server refuses DDL/DML against
|
||||||
|
-- them with error 1934 when the session has QUOTED_IDENTIFIER OFF -- which is
|
||||||
|
-- exactly what `sqlcmd` gives you by default (pass -I, or keep this header). The
|
||||||
|
-- SETs sit in their own batch so they are in force when the statements after the
|
||||||
|
-- GO parse.
|
||||||
|
SET QUOTED_IDENTIFIER ON;
|
||||||
|
SET ANSI_NULLS ON;
|
||||||
|
GO
|
||||||
|
|
||||||
|
BEGIN TRANSACTION;
|
||||||
|
IF NOT EXISTS (
|
||||||
|
SELECT * FROM [__EFMigrationsHistory]
|
||||||
|
WHERE [MigrationId] = N'20260819201210_AddNotificationsListSortIndex'
|
||||||
|
)
|
||||||
|
BEGIN
|
||||||
|
EXEC(N'CREATE INDEX [IX_Notifications_Created_NotificationId]
|
||||||
|
ON [Notifications] ([CreatedAt] DESC, [NotificationId] DESC)');
|
||||||
|
END;
|
||||||
|
|
||||||
|
IF NOT EXISTS (
|
||||||
|
SELECT * FROM [__EFMigrationsHistory]
|
||||||
|
WHERE [MigrationId] = N'20260819201210_AddNotificationsListSortIndex'
|
||||||
|
)
|
||||||
|
BEGIN
|
||||||
|
INSERT INTO [__EFMigrationsHistory] ([MigrationId], [ProductVersion])
|
||||||
|
VALUES (N'20260819201210_AddNotificationsListSortIndex', N'10.0.7');
|
||||||
|
END;
|
||||||
|
|
||||||
|
COMMIT;
|
||||||
|
GO
|
||||||
+38
@@ -1,4 +1,5 @@
|
|||||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Notifications;
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Notifications;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Notifications;
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Notifications;
|
||||||
|
|
||||||
namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||||
@@ -83,6 +84,43 @@ public interface INotificationOutboxRepository
|
|||||||
/// </returns>
|
/// </returns>
|
||||||
Task<bool> UpdateAsync(Notification n, CancellationToken cancellationToken = default);
|
Task<bool> UpdateAsync(Notification n, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Compare-and-swap variant of <see cref="UpdateAsync"/> for the operator
|
||||||
|
/// Retry/Discard one-shots: persists <paramref name="n"/>'s delivery-state
|
||||||
|
/// columns (same narrow column set as <see cref="UpdateAsync"/>) ONLY if the
|
||||||
|
/// row's current <c>Status</c> in the database is still
|
||||||
|
/// <see cref="NotificationStatus.Parked"/>. Commits internally — this call
|
||||||
|
/// is its own transaction.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <b>Why this exists (arch-review 2026-08-19 P5-4).</b> The operator retry/
|
||||||
|
/// discard handlers are read-check-write across two round trips:
|
||||||
|
/// <see cref="GetByIdAsync"/> reads the row, the handler checks its
|
||||||
|
/// in-memory <c>Status</c>, then a write persists the transition. A detached
|
||||||
|
/// <c>PipeTo</c> task can interleave with the dispatcher's own write (or a
|
||||||
|
/// second concurrent operator action) between those two round trips, so the
|
||||||
|
/// in-memory check is a fast-path rejection, not a correctness guard — a
|
||||||
|
/// blind <see cref="UpdateAsync"/> in that window can stamp a row that has
|
||||||
|
/// already moved on (e.g. the dispatcher delivered it) with the operator's
|
||||||
|
/// stale fields, silently overwriting a <c>Delivered</c> row as
|
||||||
|
/// <c>Discarded</c>. The <c>WHERE Status = Parked</c> predicate below makes
|
||||||
|
/// the <c>Parked->(Pending|Discarded)</c> transition atomic at the row
|
||||||
|
/// level, mirroring <c>SecuredWriteRepository.TryMarkApprovedAsync</c>'s
|
||||||
|
/// single-statement compare-and-swap. A row that has already progressed
|
||||||
|
/// (dispatcher won the race, or a second operator action won it) and a row
|
||||||
|
/// that has vanished (retention purge) both report zero rows affected;
|
||||||
|
/// callers cannot tell those apart from the return value alone and must
|
||||||
|
/// re-read to report which honestly.
|
||||||
|
/// </remarks>
|
||||||
|
/// <param name="n">The notification whose delivery state should be persisted, if still Parked.</param>
|
||||||
|
/// <param name="cancellationToken">Cancellation token.</param>
|
||||||
|
/// <returns>
|
||||||
|
/// <see langword="true"/> when a row with that <c>NotificationId</c> was
|
||||||
|
/// found with <c>Status == Parked</c> and its delivery state was persisted;
|
||||||
|
/// <see langword="false"/> otherwise (not found, or no longer Parked).
|
||||||
|
/// </returns>
|
||||||
|
Task<bool> TryUpdateFromParkedAsync(Notification n, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Returns a page of notifications matching <paramref name="filter"/>, ordered by
|
/// Returns a page of notifications matching <paramref name="filter"/>, ordered by
|
||||||
/// <c>CreatedAt</c> descending, together with the total matching count.
|
/// <c>CreatedAt</c> descending, together with the total matching count.
|
||||||
|
|||||||
+18
@@ -83,5 +83,23 @@ public class NotificationOutboxConfiguration : IEntityTypeConfiguration<Notifica
|
|||||||
builder.HasIndex(n => n.DeliveredAt)
|
builder.HasIndex(n => n.DeliveredAt)
|
||||||
.HasDatabaseName("IX_Notifications_Delivered")
|
.HasDatabaseName("IX_Notifications_Delivered")
|
||||||
.HasFilter("[Status] = 'Delivered'");
|
.HasFilter("[Status] = 'Delivered'");
|
||||||
|
|
||||||
|
// Created_NotificationId backs the Central UI Notification Outbox list
|
||||||
|
// page's sort key — NotificationOutboxRepository.QueryAsync orders
|
||||||
|
// (CreatedAt DESC, NotificationId DESC) (NotificationId breaks CreatedAt
|
||||||
|
// ties for deterministic OFFSET paging). None of the indexes above lead
|
||||||
|
// unconditionally on CreatedAt: Status_NextAttemptAt requires an equality
|
||||||
|
// on Status first, SourceSiteId_CreatedAt requires an equality on
|
||||||
|
// SourceSiteId first, and IX_Notifications_Delivered is filtered to
|
||||||
|
// Status='Delivered' only. So an unfiltered (or lightly filtered) page
|
||||||
|
// fell back to a full clustered-index scan + sort on every COUNT + page
|
||||||
|
// request (arch-review 2026-08-19 P5-3, register row-35 class). With both
|
||||||
|
// key columns descending, a forward scan of this index already delivers
|
||||||
|
// rows in the requested order. Exact mirror of
|
||||||
|
// SiteCallEntityTypeConfiguration's IX_SiteCalls_Created_TrackedOperationId
|
||||||
|
// (2026-08-19, arch-review P5-2).
|
||||||
|
builder.HasIndex(n => new { n.CreatedAt, n.NotificationId })
|
||||||
|
.IsDescending(true, true)
|
||||||
|
.HasDatabaseName("IX_Notifications_Created_NotificationId");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+2131
File diff suppressed because it is too large
Load Diff
+46
@@ -0,0 +1,46 @@
|
|||||||
|
using Microsoft.EntityFrameworkCore.Migrations;
|
||||||
|
|
||||||
|
#nullable disable
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Adds <c>IX_Notifications_Created_NotificationId (CreatedAt DESC,
|
||||||
|
/// NotificationId DESC)</c> — arch-review 2026-08-19 P5-3, register
|
||||||
|
/// row-35 class.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <see cref="ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories.NotificationOutboxRepository.QueryAsync"/>
|
||||||
|
/// orders <c>(CreatedAt DESC, NotificationId DESC)</c> for the Central UI
|
||||||
|
/// Notification Outbox list page. None of the existing Notifications indexes
|
||||||
|
/// lead unconditionally on <c>CreatedAt</c> — <c>IX_Notifications_Status_NextAttemptAt</c>
|
||||||
|
/// requires an equality on <c>Status</c> first, <c>IX_Notifications_SourceSiteId_CreatedAt</c>
|
||||||
|
/// requires an equality on <c>SourceSiteId</c> first, and
|
||||||
|
/// <c>IX_Notifications_Delivered</c> is filtered to <c>Status = 'Delivered'</c>
|
||||||
|
/// only — so an unfiltered (or lightly filtered) list page fell back to a full
|
||||||
|
/// clustered-index scan + sort on every COUNT + page request. With both key
|
||||||
|
/// columns descending, a forward scan of this index already delivers rows in
|
||||||
|
/// the requested order. See <c>docs/plans/sql/AddNotificationsListSortIndex.sql</c>
|
||||||
|
/// for the handwritten production-parity script.
|
||||||
|
/// </remarks>
|
||||||
|
public partial class AddNotificationsListSortIndex : Migration
|
||||||
|
{
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override void Up(MigrationBuilder migrationBuilder)
|
||||||
|
{
|
||||||
|
migrationBuilder.CreateIndex(
|
||||||
|
name: "IX_Notifications_Created_NotificationId",
|
||||||
|
table: "Notifications",
|
||||||
|
columns: new[] { "CreatedAt", "NotificationId" },
|
||||||
|
descending: new bool[0]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override void Down(MigrationBuilder migrationBuilder)
|
||||||
|
{
|
||||||
|
migrationBuilder.DropIndex(
|
||||||
|
name: "IX_Notifications_Created_NotificationId",
|
||||||
|
table: "Notifications");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+4
@@ -913,6 +913,10 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
|
|||||||
.HasDatabaseName("IX_Notifications_Delivered")
|
.HasDatabaseName("IX_Notifications_Delivered")
|
||||||
.HasFilter("[Status] = 'Delivered'");
|
.HasFilter("[Status] = 'Delivered'");
|
||||||
|
|
||||||
|
b.HasIndex("CreatedAt", "NotificationId")
|
||||||
|
.IsDescending()
|
||||||
|
.HasDatabaseName("IX_Notifications_Created_NotificationId");
|
||||||
|
|
||||||
b.HasIndex("SourceSiteId", "CreatedAt");
|
b.HasIndex("SourceSiteId", "CreatedAt");
|
||||||
|
|
||||||
b.HasIndex("Status", "NextAttemptAt");
|
b.HasIndex("Status", "NextAttemptAt");
|
||||||
|
|||||||
+42
@@ -205,6 +205,48 @@ VALUES
|
|||||||
return rowsAffected > 0;
|
return rowsAffected > 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<bool> TryUpdateFromParkedAsync(Notification n, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(n);
|
||||||
|
|
||||||
|
// Same seven mutable delivery-state columns as UpdateAsync, but the WHERE
|
||||||
|
// clause below ALSO requires the row's current Status to still be Parked
|
||||||
|
// -- a single-statement compare-and-swap (arch-review 2026-08-19 P5-4).
|
||||||
|
// Two round trips (GetByIdAsync then this write) with a status check in
|
||||||
|
// between is not a correctness guard by itself: a detached PipeTo task
|
||||||
|
// can race the dispatcher's own write (or a second concurrent operator
|
||||||
|
// action) in that window. Mirrors SecuredWriteRepository.TryMarkApprovedAsync's
|
||||||
|
// conditional ExecuteUpdate/ExecuteSql pattern, expressed via LINQ
|
||||||
|
// ExecuteUpdateAsync (like UpdateAsync above) so this stays portable
|
||||||
|
// across the SQL Server production provider and the SQLite provider the
|
||||||
|
// repository tests run against.
|
||||||
|
var status = n.Status;
|
||||||
|
var retryCount = n.RetryCount;
|
||||||
|
var lastError = n.LastError;
|
||||||
|
var resolvedTargets = n.ResolvedTargets;
|
||||||
|
var lastAttemptAt = n.LastAttemptAt;
|
||||||
|
var nextAttemptAt = n.NextAttemptAt;
|
||||||
|
var deliveredAt = n.DeliveredAt;
|
||||||
|
|
||||||
|
var notificationId = n.NotificationId;
|
||||||
|
|
||||||
|
var rowsAffected = await _context.Notifications
|
||||||
|
.Where(row => row.NotificationId == notificationId && row.Status == NotificationStatus.Parked)
|
||||||
|
.ExecuteUpdateAsync(
|
||||||
|
setters => setters
|
||||||
|
.SetProperty(row => row.Status, status)
|
||||||
|
.SetProperty(row => row.RetryCount, retryCount)
|
||||||
|
.SetProperty(row => row.LastError, lastError)
|
||||||
|
.SetProperty(row => row.ResolvedTargets, resolvedTargets)
|
||||||
|
.SetProperty(row => row.LastAttemptAt, lastAttemptAt)
|
||||||
|
.SetProperty(row => row.NextAttemptAt, nextAttemptAt)
|
||||||
|
.SetProperty(row => row.DeliveredAt, deliveredAt),
|
||||||
|
cancellationToken);
|
||||||
|
|
||||||
|
return rowsAffected > 0;
|
||||||
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public async Task<Notification?> GetByIdAsync(string notificationId, CancellationToken cancellationToken = default)
|
public async Task<Notification?> GetByIdAsync(string notificationId, CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -1109,14 +1109,25 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
|||||||
notification.NextAttemptAt = null;
|
notification.NextAttemptAt = null;
|
||||||
notification.LastError = null;
|
notification.LastError = null;
|
||||||
|
|
||||||
// Zero rows updated means the row was purged between the read above and
|
// Compare-and-swap on Status == Parked (arch-review 2026-08-19 P5-4). The
|
||||||
// this write. Answer the operator honestly instead of reporting a
|
// read above and this write are two separate round trips; a detached
|
||||||
// re-queue that never happened — and emit no un-park audit row, because
|
// PipeTo task can race the dispatcher's own write (or a second
|
||||||
// there is nothing to attribute it to.
|
// concurrent operator action) in between, so the in-memory check above
|
||||||
if (!await repository.UpdateAsync(notification))
|
// is a fast-path rejection, not the correctness guard — a blind
|
||||||
|
// UpdateAsync here could stamp a row that has already moved on (e.g. the
|
||||||
|
// dispatcher delivered it) back to Pending with stale fields. A lost CAS
|
||||||
|
// means either the row vanished (purge) or it progressed past Parked
|
||||||
|
// under us; re-read to report which honestly instead of silently
|
||||||
|
// overwriting whatever is there now, and emit no un-park audit row in
|
||||||
|
// either case, because there is nothing to attribute it to.
|
||||||
|
if (!await repository.TryUpdateFromParkedAsync(notification))
|
||||||
{
|
{
|
||||||
|
var current = await repository.GetByIdAsync(request.NotificationId);
|
||||||
return new RetryNotificationResponse(
|
return new RetryNotificationResponse(
|
||||||
request.CorrelationId, Success: false, ErrorMessage: "notification not found");
|
request.CorrelationId, Success: false,
|
||||||
|
ErrorMessage: current is null
|
||||||
|
? "notification not found"
|
||||||
|
: $"notification already progressed to {current.Status}; retry not applied");
|
||||||
}
|
}
|
||||||
|
|
||||||
// Operator re-queued a parked notification. Emit a Submitted NotifyDeliver
|
// Operator re-queued a parked notification. Emit a Submitted NotifyDeliver
|
||||||
@@ -1190,13 +1201,22 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
|
|||||||
|
|
||||||
notification.Status = NotificationStatus.Discarded;
|
notification.Status = NotificationStatus.Discarded;
|
||||||
|
|
||||||
// Same vanished-row honesty as the retry path: a purge between the read
|
// Same compare-and-swap on Status == Parked as the retry path (arch-review
|
||||||
// and the write leaves nothing to discard, so report not-found rather
|
// 2026-08-19 P5-4): a purge between the read and the write leaves nothing
|
||||||
// than success (and skip the terminal audit row below).
|
// to discard, and a dispatcher/other-operator write landing in that same
|
||||||
if (!await repository.UpdateAsync(notification))
|
// window (e.g. a delivery succeeding) must not be silently clobbered by a
|
||||||
|
// blind write of this stale in-memory copy. Re-read on a lost CAS to
|
||||||
|
// report which case it was, and skip the terminal audit row below either
|
||||||
|
// way — there is nothing to attribute a discard of a row that was never
|
||||||
|
// actually discarded.
|
||||||
|
if (!await repository.TryUpdateFromParkedAsync(notification))
|
||||||
{
|
{
|
||||||
|
var current = await repository.GetByIdAsync(request.NotificationId);
|
||||||
return new DiscardNotificationResponse(
|
return new DiscardNotificationResponse(
|
||||||
request.CorrelationId, Success: false, ErrorMessage: "notification not found");
|
request.CorrelationId, Success: false,
|
||||||
|
ErrorMessage: current is null
|
||||||
|
? "notification not found"
|
||||||
|
: $"notification already progressed to {current.Status}; discard not applied");
|
||||||
}
|
}
|
||||||
|
|
||||||
// A manual discard is the OTHER code path that produces
|
// A manual discard is the OTHER code path that produces
|
||||||
|
|||||||
@@ -573,6 +573,154 @@ public class NotificationOutboxRepositoryTests : IDisposable
|
|||||||
Assert.False(await _repository.UpdateAsync(loaded));
|
Assert.False(await _repository.UpdateAsync(loaded));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task TryUpdateFromParkedAsync_StatusIsParked_UpdatesAndReturnsTrue()
|
||||||
|
{
|
||||||
|
var id = Guid.NewGuid().ToString();
|
||||||
|
_context.Notifications.Add(MakeNotification(id, NotificationStatus.Parked));
|
||||||
|
await _context.SaveChangesAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
|
||||||
|
var loaded = await _repository.GetByIdAsync(id);
|
||||||
|
loaded!.Status = NotificationStatus.Pending;
|
||||||
|
loaded.RetryCount = 0;
|
||||||
|
loaded.NextAttemptAt = null;
|
||||||
|
loaded.LastError = null;
|
||||||
|
Assert.True(await _repository.TryUpdateFromParkedAsync(loaded));
|
||||||
|
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
var reloaded = await _context.Notifications.FindAsync(id);
|
||||||
|
Assert.Equal(NotificationStatus.Pending, reloaded!.Status);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Direct CAS-miss: the row is currently Delivered (never Parked in this
|
||||||
|
/// scenario), so the conditional WHERE Status == Parked matches nothing and
|
||||||
|
/// the row is left completely untouched — arch-review 2026-08-19 P5-4.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task TryUpdateFromParkedAsync_StatusIsNotParked_ReturnsFalse_AndLeavesRowUnchanged()
|
||||||
|
{
|
||||||
|
var id = Guid.NewGuid().ToString();
|
||||||
|
var deliveredAt = new DateTimeOffset(2026, 5, 19, 9, 0, 0, TimeSpan.Zero);
|
||||||
|
_context.Notifications.Add(MakeNotification(
|
||||||
|
id, NotificationStatus.Delivered, deliveredAt: deliveredAt));
|
||||||
|
await _context.SaveChangesAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
|
||||||
|
var staleDiscard = MakeNotification(id, NotificationStatus.Discarded);
|
||||||
|
Assert.False(await _repository.TryUpdateFromParkedAsync(staleDiscard));
|
||||||
|
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
var reloaded = await _context.Notifications.FindAsync(id);
|
||||||
|
Assert.Equal(NotificationStatus.Delivered, reloaded!.Status);
|
||||||
|
Assert.Equal(deliveredAt, reloaded.DeliveredAt);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task TryUpdateFromParkedAsync_RowPurgedBeforeWrite_ReturnsFalse()
|
||||||
|
{
|
||||||
|
var id = Guid.NewGuid().ToString();
|
||||||
|
_context.Notifications.Add(MakeNotification(id, NotificationStatus.Parked));
|
||||||
|
await _context.SaveChangesAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
|
||||||
|
var loaded = await _repository.GetByIdAsync(id);
|
||||||
|
Assert.NotNull(loaded);
|
||||||
|
|
||||||
|
await _context.Notifications.Where(n => n.NotificationId == id).ExecuteDeleteAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
|
||||||
|
loaded!.Status = NotificationStatus.Discarded;
|
||||||
|
Assert.False(await _repository.TryUpdateFromParkedAsync(loaded));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Deterministic (non-sleep) proof of the arch-review 2026-08-19 P5-4 race:
|
||||||
|
/// the operator reads a Parked row, but before its write lands the
|
||||||
|
/// dispatcher (or a concurrent second operator action) has already moved
|
||||||
|
/// the row to Delivered. <see cref="TryUpdateFromParkedAsync"/>'s CAS must
|
||||||
|
/// refuse the stale write and leave the Delivered row exactly as it is.
|
||||||
|
/// The companion fact below proves the general-purpose
|
||||||
|
/// <see cref="NotificationOutboxRepository.UpdateAsync"/> — the method the
|
||||||
|
/// operator path used PRE-FIX — has no such guard, which is the mechanism
|
||||||
|
/// the finding describes.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task TryUpdateFromParkedAsync_ConcurrentlyDelivered_ReturnsFalse_AndDoesNotOverwrite()
|
||||||
|
{
|
||||||
|
var id = Guid.NewGuid().ToString();
|
||||||
|
_context.Notifications.Add(MakeNotification(id, NotificationStatus.Parked));
|
||||||
|
await _context.SaveChangesAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
|
||||||
|
// The operator's read (this snapshot is carried forward stale below).
|
||||||
|
var operatorRead = await _repository.GetByIdAsync(id);
|
||||||
|
Assert.NotNull(operatorRead);
|
||||||
|
|
||||||
|
// The dispatcher's own write lands first, in the window between the
|
||||||
|
// operator's read and write.
|
||||||
|
var deliveredAt = new DateTimeOffset(2026, 5, 19, 9, 0, 0, TimeSpan.Zero);
|
||||||
|
var dispatcherRead = await _repository.GetByIdAsync(id);
|
||||||
|
dispatcherRead!.Status = NotificationStatus.Delivered;
|
||||||
|
dispatcherRead.DeliveredAt = deliveredAt;
|
||||||
|
Assert.True(await _repository.UpdateAsync(dispatcherRead));
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
|
||||||
|
// The operator's stale in-memory copy still says Parked; it attempts a Discard.
|
||||||
|
operatorRead!.Status = NotificationStatus.Discarded;
|
||||||
|
Assert.False(await _repository.TryUpdateFromParkedAsync(operatorRead));
|
||||||
|
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
var reloaded = await _context.Notifications.FindAsync(id);
|
||||||
|
Assert.Equal(NotificationStatus.Delivered, reloaded!.Status);
|
||||||
|
Assert.Equal(deliveredAt, reloaded.DeliveredAt);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Companion to <see cref="TryUpdateFromParkedAsync_ConcurrentlyDelivered_ReturnsFalse_AndDoesNotOverwrite"/>:
|
||||||
|
/// runs the IDENTICAL interleaving through the unconditional
|
||||||
|
/// <see cref="NotificationOutboxRepository.UpdateAsync"/> instead — the
|
||||||
|
/// method the operator Retry/Discard path called pre-fix — and shows it
|
||||||
|
/// silently clobbers the Delivered row. This documents the exact defect
|
||||||
|
/// arch-review 2026-08-19 P5-4 fixed by routing the operator path through
|
||||||
|
/// <see cref="TryUpdateFromParkedAsync"/>: UpdateAsync itself is unchanged
|
||||||
|
/// (the dispatcher still needs an unconditional write), so this scenario
|
||||||
|
/// remains "true" by design — it is the operator actor's call-site choice
|
||||||
|
/// that was the bug, proven fixed by the actor-level tests in
|
||||||
|
/// NotificationOutbox.Tests (Discard_ConcurrentlyDelivered_...).
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public async Task UpdateAsync_Unconditional_WouldOverwriteConcurrentlyDeliveredRow_DemonstratingP5_4Mechanism()
|
||||||
|
{
|
||||||
|
var id = Guid.NewGuid().ToString();
|
||||||
|
_context.Notifications.Add(MakeNotification(id, NotificationStatus.Parked));
|
||||||
|
await _context.SaveChangesAsync();
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
|
||||||
|
var operatorRead = await _repository.GetByIdAsync(id);
|
||||||
|
Assert.NotNull(operatorRead);
|
||||||
|
|
||||||
|
var deliveredAt = new DateTimeOffset(2026, 5, 19, 9, 0, 0, TimeSpan.Zero);
|
||||||
|
var dispatcherRead = await _repository.GetByIdAsync(id);
|
||||||
|
dispatcherRead!.Status = NotificationStatus.Delivered;
|
||||||
|
dispatcherRead.DeliveredAt = deliveredAt;
|
||||||
|
Assert.True(await _repository.UpdateAsync(dispatcherRead));
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
|
||||||
|
operatorRead!.Status = NotificationStatus.Discarded;
|
||||||
|
// The unconditional UPDATE has no status guard, so it "succeeds" here —
|
||||||
|
// this is the bug. It clobbers the Delivered row to Discarded and drops
|
||||||
|
// DeliveredAt, which is exactly the silent-overwrite hazard P5-4 closes
|
||||||
|
// by moving the operator path to TryUpdateFromParkedAsync.
|
||||||
|
Assert.True(await _repository.UpdateAsync(operatorRead));
|
||||||
|
|
||||||
|
_context.ChangeTracker.Clear();
|
||||||
|
var reloaded = await _context.Notifications.FindAsync(id);
|
||||||
|
Assert.Equal(NotificationStatus.Discarded, reloaded!.Status);
|
||||||
|
Assert.Null(reloaded.DeliveredAt);
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task QueryAsync_AppliesFilters_OrdersByCreatedAtDescending_AndPaginates()
|
public async Task QueryAsync_AppliesFilters_OrdersByCreatedAtDescending_AndPaginates()
|
||||||
{
|
{
|
||||||
|
|||||||
+84
-12
@@ -258,7 +258,7 @@ public class NotificationOutboxActorQueryTests : TestKit
|
|||||||
Assert.True(response.Success);
|
Assert.True(response.Success);
|
||||||
Assert.Null(response.ErrorMessage);
|
Assert.Null(response.ErrorMessage);
|
||||||
|
|
||||||
_repository.Received(1).UpdateAsync(
|
_repository.Received(1).TryUpdateFromParkedAsync(
|
||||||
Arg.Is<Notification>(n =>
|
Arg.Is<Notification>(n =>
|
||||||
n.NotificationId == row.NotificationId &&
|
n.NotificationId == row.NotificationId &&
|
||||||
n.Status == NotificationStatus.Pending &&
|
n.Status == NotificationStatus.Pending &&
|
||||||
@@ -280,7 +280,7 @@ public class NotificationOutboxActorQueryTests : TestKit
|
|||||||
var response = ExpectMsg<RetryNotificationResponse>();
|
var response = ExpectMsg<RetryNotificationResponse>();
|
||||||
Assert.False(response.Success);
|
Assert.False(response.Success);
|
||||||
Assert.NotNull(response.ErrorMessage);
|
Assert.NotNull(response.ErrorMessage);
|
||||||
_repository.DidNotReceive().UpdateAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>());
|
_repository.DidNotReceive().TryUpdateFromParkedAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>());
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
@@ -300,17 +300,19 @@ public class NotificationOutboxActorQueryTests : TestKit
|
|||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// The row is read successfully but the retention purge deletes it before the
|
/// The row is read successfully but the retention purge deletes it before the
|
||||||
/// write lands, so the targeted UPDATE matches nothing. The operator must be
|
/// write lands, so the CAS write matches nothing. The re-read that follows a
|
||||||
/// told the notification is gone — reporting a re-queue that never happened
|
/// lost CAS (arch-review 2026-08-19 P5-4) finds nothing either, and the
|
||||||
/// is what the ExecuteUpdate rewrite silently introduced (the predecessor
|
/// operator must be told the notification is gone — reporting a re-queue
|
||||||
/// threw DbUpdateConcurrencyException here).
|
/// that never happened is what a blind ExecuteUpdate would silently do (the
|
||||||
|
/// pre-ExecuteUpdate predecessor threw DbUpdateConcurrencyException here).
|
||||||
/// </summary>
|
/// </summary>
|
||||||
[Fact]
|
[Fact]
|
||||||
public void Retry_RowPurgedBeforeWrite_RepliesNotFound()
|
public void Retry_RowPurgedBeforeWrite_RepliesNotFound()
|
||||||
{
|
{
|
||||||
var row = MakeNotification(status: NotificationStatus.Parked, retryCount: 10, lastError: "gave up");
|
var row = MakeNotification(status: NotificationStatus.Parked, retryCount: 10, lastError: "gave up");
|
||||||
_repository.GetByIdAsync(row.NotificationId, Arg.Any<CancellationToken>()).Returns(row);
|
_repository.GetByIdAsync(row.NotificationId, Arg.Any<CancellationToken>())
|
||||||
_repository.UpdateAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>()).Returns(false);
|
.Returns(row, (Notification?)null);
|
||||||
|
_repository.TryUpdateFromParkedAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>()).Returns(false);
|
||||||
var actor = CreateActor();
|
var actor = CreateActor();
|
||||||
|
|
||||||
actor.Tell(new RetryNotificationRequest("corr-vanished-retry", row.NotificationId), TestActor);
|
actor.Tell(new RetryNotificationRequest("corr-vanished-retry", row.NotificationId), TestActor);
|
||||||
@@ -327,8 +329,9 @@ public class NotificationOutboxActorQueryTests : TestKit
|
|||||||
public void Discard_RowPurgedBeforeWrite_RepliesNotFound()
|
public void Discard_RowPurgedBeforeWrite_RepliesNotFound()
|
||||||
{
|
{
|
||||||
var row = MakeNotification(status: NotificationStatus.Parked);
|
var row = MakeNotification(status: NotificationStatus.Parked);
|
||||||
_repository.GetByIdAsync(row.NotificationId, Arg.Any<CancellationToken>()).Returns(row);
|
_repository.GetByIdAsync(row.NotificationId, Arg.Any<CancellationToken>())
|
||||||
_repository.UpdateAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>()).Returns(false);
|
.Returns(row, (Notification?)null);
|
||||||
|
_repository.TryUpdateFromParkedAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>()).Returns(false);
|
||||||
var actor = CreateActor();
|
var actor = CreateActor();
|
||||||
|
|
||||||
actor.Tell(new DiscardNotificationRequest("corr-vanished-discard", row.NotificationId), TestActor);
|
actor.Tell(new DiscardNotificationRequest("corr-vanished-discard", row.NotificationId), TestActor);
|
||||||
@@ -340,6 +343,75 @@ public class NotificationOutboxActorQueryTests : TestKit
|
|||||||
Assert.Contains("not found", response.ErrorMessage);
|
Assert.Contains("not found", response.ErrorMessage);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// PROOF OF FAILURE for arch-review 2026-08-19 P5-4. Models the race the
|
||||||
|
/// finding describes: the operator's read sees Parked, but by the time the
|
||||||
|
/// write lands the dispatcher has already delivered the row (a detached
|
||||||
|
/// PipeTo task interleaving with the dispatcher's own write). The mechanism
|
||||||
|
/// is proven deterministically via a mocked call sequence, not a wall-clock
|
||||||
|
/// sleep (per the "no new sleep-then-assert" constraint) — the first
|
||||||
|
/// <see cref="INotificationOutboxRepository.GetByIdAsync"/> call (the
|
||||||
|
/// operator's read) returns the Parked row, the CAS write reports a lost
|
||||||
|
/// race, and the SECOND <c>GetByIdAsync</c> call (the fix's re-read used to
|
||||||
|
/// build an honest reply) returns the row now Delivered.
|
||||||
|
/// <para>
|
||||||
|
/// Against the pre-fix actor — which calls the unconditional
|
||||||
|
/// <c>UpdateAsync</c> exactly once and never re-reads — this test fails:
|
||||||
|
/// the single-read code path only ever consumes the first queued value
|
||||||
|
/// (Parked), the in-memory check passes, <c>UpdateAsync</c> is stubbed
|
||||||
|
/// <c>Healthy()</c> (returns true), and the response comes back
|
||||||
|
/// <c>Success = true</c> — i.e. the Delivered row gets silently clobbered
|
||||||
|
/// to Discarded. Post-fix, the CAS is lost, the actor re-reads and reports
|
||||||
|
/// "already progressed to Delivered" instead of overwriting it.
|
||||||
|
/// </para>
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void Discard_ConcurrentlyDelivered_RepliesAlreadyProgressed_DoesNotOverwrite()
|
||||||
|
{
|
||||||
|
var parkedRow = MakeNotification(status: NotificationStatus.Parked, retryCount: 5, lastError: "gave up");
|
||||||
|
var deliveredRow = MakeNotification(status: NotificationStatus.Delivered);
|
||||||
|
_repository.GetByIdAsync(parkedRow.NotificationId, Arg.Any<CancellationToken>())
|
||||||
|
.Returns(parkedRow, deliveredRow);
|
||||||
|
_repository.TryUpdateFromParkedAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>()).Returns(false);
|
||||||
|
var actor = CreateActor();
|
||||||
|
|
||||||
|
actor.Tell(new DiscardNotificationRequest("corr-race-discard", parkedRow.NotificationId), TestActor);
|
||||||
|
|
||||||
|
var response = ExpectMsg<DiscardNotificationResponse>();
|
||||||
|
Assert.Equal("corr-race-discard", response.CorrelationId);
|
||||||
|
Assert.False(response.Success);
|
||||||
|
Assert.NotNull(response.ErrorMessage);
|
||||||
|
Assert.Contains("already progressed", response.ErrorMessage);
|
||||||
|
Assert.Contains("Delivered", response.ErrorMessage);
|
||||||
|
|
||||||
|
// The critical invariant: the old unconditional UpdateAsync must NEVER
|
||||||
|
// be reached by the operator path — that call is what silently
|
||||||
|
// clobbered the Delivered row pre-fix.
|
||||||
|
_repository.DidNotReceive().UpdateAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Retry half of <see cref="Discard_ConcurrentlyDelivered_RepliesAlreadyProgressed_DoesNotOverwrite"/>.</summary>
|
||||||
|
[Fact]
|
||||||
|
public void Retry_ConcurrentlyDelivered_RepliesAlreadyProgressed_DoesNotOverwrite()
|
||||||
|
{
|
||||||
|
var parkedRow = MakeNotification(status: NotificationStatus.Parked, retryCount: 5, lastError: "gave up");
|
||||||
|
var deliveredRow = MakeNotification(status: NotificationStatus.Delivered);
|
||||||
|
_repository.GetByIdAsync(parkedRow.NotificationId, Arg.Any<CancellationToken>())
|
||||||
|
.Returns(parkedRow, deliveredRow);
|
||||||
|
_repository.TryUpdateFromParkedAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>()).Returns(false);
|
||||||
|
var actor = CreateActor();
|
||||||
|
|
||||||
|
actor.Tell(new RetryNotificationRequest("corr-race-retry", parkedRow.NotificationId), TestActor);
|
||||||
|
|
||||||
|
var response = ExpectMsg<RetryNotificationResponse>();
|
||||||
|
Assert.Equal("corr-race-retry", response.CorrelationId);
|
||||||
|
Assert.False(response.Success);
|
||||||
|
Assert.NotNull(response.ErrorMessage);
|
||||||
|
Assert.Contains("already progressed", response.ErrorMessage);
|
||||||
|
Assert.Contains("Delivered", response.ErrorMessage);
|
||||||
|
_repository.DidNotReceive().UpdateAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>());
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void Discard_ParkedNotification_MarksDiscarded_AndSucceeds()
|
public void Discard_ParkedNotification_MarksDiscarded_AndSucceeds()
|
||||||
{
|
{
|
||||||
@@ -353,7 +425,7 @@ public class NotificationOutboxActorQueryTests : TestKit
|
|||||||
Assert.Equal("corr-8", response.CorrelationId);
|
Assert.Equal("corr-8", response.CorrelationId);
|
||||||
Assert.True(response.Success);
|
Assert.True(response.Success);
|
||||||
|
|
||||||
_repository.Received(1).UpdateAsync(
|
_repository.Received(1).TryUpdateFromParkedAsync(
|
||||||
Arg.Is<Notification>(n => n.Status == NotificationStatus.Discarded),
|
Arg.Is<Notification>(n => n.Status == NotificationStatus.Discarded),
|
||||||
Arg.Any<CancellationToken>());
|
Arg.Any<CancellationToken>());
|
||||||
}
|
}
|
||||||
@@ -370,7 +442,7 @@ public class NotificationOutboxActorQueryTests : TestKit
|
|||||||
var response = ExpectMsg<DiscardNotificationResponse>();
|
var response = ExpectMsg<DiscardNotificationResponse>();
|
||||||
Assert.False(response.Success);
|
Assert.False(response.Success);
|
||||||
Assert.NotNull(response.ErrorMessage);
|
Assert.NotNull(response.ErrorMessage);
|
||||||
_repository.DidNotReceive().UpdateAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>());
|
_repository.DidNotReceive().TryUpdateFromParkedAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>());
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
+1
-1
@@ -147,7 +147,7 @@ public class NotificationOutboxActorRetryEmissionTests : TestKit
|
|||||||
ExpectMsg<RetryNotificationResponse>(r => r.Success);
|
ExpectMsg<RetryNotificationResponse>(r => r.Success);
|
||||||
AwaitAssert(() =>
|
AwaitAssert(() =>
|
||||||
{
|
{
|
||||||
_outboxRepository.Received(1).UpdateAsync(
|
_outboxRepository.Received(1).TryUpdateFromParkedAsync(
|
||||||
Arg.Is<Notification>(n => n.Status == NotificationStatus.Pending),
|
Arg.Is<Notification>(n => n.Status == NotificationStatus.Pending),
|
||||||
Arg.Any<CancellationToken>());
|
Arg.Any<CancellationToken>());
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -8,14 +8,16 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests;
|
|||||||
/// Factory for the outbox repository substitute used across these tests.
|
/// Factory for the outbox repository substitute used across these tests.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
/// <see cref="INotificationOutboxRepository.UpdateAsync"/> returns whether the
|
/// <see cref="INotificationOutboxRepository.UpdateAsync"/> and
|
||||||
/// targeted UPDATE actually matched a row — <c>false</c> is the "this
|
/// <see cref="INotificationOutboxRepository.TryUpdateFromParkedAsync"/> both
|
||||||
/// notification no longer exists" signal the operator retry/discard handlers
|
/// return whether the targeted UPDATE actually matched a row — <c>false</c> is
|
||||||
/// turn into a not-found reply. NSubstitute's default for <c>Task<bool></c>
|
/// the "this notification no longer exists (or, for the CAS variant, is no
|
||||||
/// is <c>false</c>, i.e. the VANISHED-row answer, so an unconfigured substitute
|
/// longer Parked)" signal the operator retry/discard handlers turn into a
|
||||||
/// would silently put every test on the failure path. Tests that want a healthy
|
/// not-found / already-progressed reply. NSubstitute's default for
|
||||||
/// store take one from here; the vanished-row tests configure
|
/// <c>Task<bool></c> is <c>false</c>, i.e. the failure answer, so an
|
||||||
/// <c>Returns(false)</c> for themselves.
|
/// unconfigured substitute would silently put every test on the failure path.
|
||||||
|
/// Tests that want a healthy store take one from here; the vanished-row and
|
||||||
|
/// lost-CAS tests configure <c>Returns(false)</c> for themselves.
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
internal static class OutboxRepositorySubstitute
|
internal static class OutboxRepositorySubstitute
|
||||||
{
|
{
|
||||||
@@ -26,6 +28,9 @@ internal static class OutboxRepositorySubstitute
|
|||||||
repository
|
repository
|
||||||
.UpdateAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>())
|
.UpdateAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>())
|
||||||
.Returns(true);
|
.Returns(true);
|
||||||
|
repository
|
||||||
|
.TryUpdateFromParkedAsync(Arg.Any<Notification>(), Arg.Any<CancellationToken>())
|
||||||
|
.Returns(true);
|
||||||
return repository;
|
return repository;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user