diff --git a/docs/plans/sql/AddAuditLogAndNotificationCoveringIndexes.sql b/docs/plans/sql/AddAuditLogAndNotificationCoveringIndexes.sql index c051c106..db6084d4 100644 --- a/docs/plans/sql/AddAuditLogAndNotificationCoveringIndexes.sql +++ b/docs/plans/sql/AddAuditLogAndNotificationCoveringIndexes.sql @@ -1,4 +1,14 @@ -BEGIN TRANSACTION; +-- Run with QUOTED_IDENTIFIER ON. SiteCalls / Notifications / AuditLog all carry +-- filtered indexes (and AuditLog persisted computed columns), and SQL Server +-- refuses BOTH the filtered-index DDL below and any later DML on those tables +-- 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'20260814235335_AddAuditLogAndNotificationCoveringIndexes' diff --git a/docs/plans/sql/AddSiteCallsNonTerminalIndex.sql b/docs/plans/sql/AddSiteCallsNonTerminalIndex.sql index a04634a7..c2ec20ed 100644 --- a/docs/plans/sql/AddSiteCallsNonTerminalIndex.sql +++ b/docs/plans/sql/AddSiteCallsNonTerminalIndex.sql @@ -1,4 +1,14 @@ -BEGIN TRANSACTION; +-- Run with QUOTED_IDENTIFIER ON. SiteCalls / Notifications / AuditLog all carry +-- filtered indexes (and AuditLog persisted computed columns), and SQL Server +-- refuses BOTH the filtered-index DDL below and any later DML on those tables +-- 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'20260709110614_AddSiteCallsNonTerminalIndex' diff --git a/docs/plans/sql/AddSiteCallsTerminalIndex.sql b/docs/plans/sql/AddSiteCallsTerminalIndex.sql index 3486a30e..48405510 100644 --- a/docs/plans/sql/AddSiteCallsTerminalIndex.sql +++ b/docs/plans/sql/AddSiteCallsTerminalIndex.sql @@ -1,4 +1,14 @@ -BEGIN TRANSACTION; +-- Run with QUOTED_IDENTIFIER ON. SiteCalls / Notifications / AuditLog all carry +-- filtered indexes (and AuditLog persisted computed columns), and SQL Server +-- refuses BOTH the filtered-index DDL below and any later DML on those tables +-- 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'20260713142234_AddSiteCallsTerminalIndex' diff --git a/docs/plans/sql/AlignAuditLogEventIdUniqueness.sql b/docs/plans/sql/AlignAuditLogEventIdUniqueness.sql index 90b3003f..7c97570c 100644 --- a/docs/plans/sql/AlignAuditLogEventIdUniqueness.sql +++ b/docs/plans/sql/AlignAuditLogEventIdUniqueness.sql @@ -1,4 +1,14 @@ -BEGIN TRANSACTION; +-- Run with QUOTED_IDENTIFIER ON. SiteCalls / Notifications / AuditLog all carry +-- filtered indexes (and AuditLog persisted computed columns), and SQL Server +-- refuses BOTH the filtered-index DDL below and any later DML on those tables +-- 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'20260815004957_AlignAuditLogEventIdUniqueness' diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index b9364d22..3b9a9be1 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -303,6 +303,24 @@ per-row / per-entry path — so the documented invariant that one bad row cannot sink the rest of the batch still holds, it is simply no longer paid for on the healthy path. +The set-based path binds its string parameters at the VALUE's own length, never +at the column width. Declaring the width makes the client TRUNCATE an over-long +value at bind time and commit the shortened row — silent mutilation in an +append-only store, with no `PayloadTruncated` flag to admit it, and inconsistent +with the per-row and reconciliation paths, which send the value in full and let +the server reject it. Length enforcement belongs to the server on every path: +reject everywhere, truncate nowhere. (Deliberate, policy-driven truncation of +`RequestSummary`/`ResponseSummary` under the payload caps is a different thing +entirely — it happens before the write and always sets `PayloadTruncated`.) + +The per-row fallback runs on its OWN short cancellation budget rather than the +batch's. Sharing it meant that a batch failing *because* the ingest budget +expired handed every fallback insert an already-cancelled token: N instant +failures, N counter bumps, nothing accepted — the fallback defeated at exactly +the moment it was needed. A blown budget is also counted ONCE for the batch +rather than once per row, so the health counter reads as one timeout instead of +a burst of write failures. + **Timeout ladder.** The ingest budget is deliberately the smallest on the path: the site's Ask and the central gRPC handler's Ask are both 30 s, the actor's own database budget is 20 s and the per-statement SQL timeout is 15 s. Before diff --git a/docs/requirements/Component-ConfigurationDatabase.md b/docs/requirements/Component-ConfigurationDatabase.md index c246cd8c..5aa52d8f 100644 --- a/docs/requirements/Component-ConfigurationDatabase.md +++ b/docs/requirements/Component-ConfigurationDatabase.md @@ -379,6 +379,22 @@ dotnet ef migrations script FromMigration ToMigration --output migration.sql Generated scripts are idempotent — they can be safely re-run without causing errors or duplicate changes. +### Running a Script: QUOTED_IDENTIFIER + +`SiteCalls`, `Notifications` and `AuditLog` all carry **filtered indexes**, and `AuditLog` additionally carries persisted computed columns. SQL Server refuses both the filtered-index DDL and any subsequent `INSERT`/`UPDATE`/`DELETE` on such a table with **error 1934** unless the session has `QUOTED_IDENTIFIER ON`. + +`sqlcmd` defaults that option **OFF** — including the `docker exec … /opt/mssql-tools18/bin/sqlcmd` invocation used throughout the test-infra docs — so a script that applies cleanly in SSMS (which defaults ON) fails there. Two mitigations, both in place: + +- Every checked-in script under `docs/plans/sql/` begins with a `SET QUOTED_IDENTIFIER ON; SET ANSI_NULLS ON; GO` header, in its own batch so the setting is in force when the following batch parses. +- Run `sqlcmd` with **`-I`** (enable quoted identifiers) regardless — it is the correct default for every script this repo generates, and it protects hand-written one-off statements that carry no header: + +```bash +docker exec -i scadabridge-mssql /opt/mssql-tools18/bin/sqlcmd \ + -S localhost -U sa -P 'ScadaBridge_Dev1#' -C -I -d ScadaBridgeConfig -i /path/to/script.sql +``` + +The application itself is unaffected: SqlClient sets `QUOTED_IDENTIFIER ON` on every connection, so EF Core migrations and all runtime DML already run with it on. + --- ## Seed Data diff --git a/docs/requirements/Component-NotificationOutbox.md b/docs/requirements/Component-NotificationOutbox.md index 6e5293eb..718dd979 100644 --- a/docs/requirements/Component-NotificationOutbox.md +++ b/docs/requirements/Component-NotificationOutbox.md @@ -167,6 +167,8 @@ A notification is **stuck** if it is `Pending` or `Retrying` and older than a co - **Health Monitoring dashboard** — headline KPI tiles: queue depth, stuck count, parked count. These are central-computed and are not part of the site health report. The site S&F notification backlog remains a separate site health metric covering the site→central leg. - **Central UI "Notification Outbox" page** — KPI tiles plus a queryable notification list: filter by status, type, source site, list, and time range; a stuck-only toggle; keyword search on subject. Parked notifications offer **Retry** (→ `Pending`, reset `RetryCount` / `NextAttemptAt`) and **Discard** (→ `Discarded`) actions. Stuck rows are badged. +Both operator actions are **read-then-write against a row the daily retention purge may delete in between**, so the write reports whether it matched a row and the actor answers `Success: false` / `"notification not found"` when it did not — never a success against a notification that no longer exists (and no audit row is emitted for the action that did not happen). The dispatcher's own delivery-state write takes the opposite stance on the same signal: the delivery has already happened, there is nothing left to retry, so a vanished row is logged as a warning and the audit rows stand as the durable record. + ## Configuration The component is configured via `NotificationOutboxOptions`, bound from an `appsettings.json` section on the central host (Options pattern): diff --git a/docs/requirements/Component-SiteCallAudit.md b/docs/requirements/Component-SiteCallAudit.md index 58b2388f..ee2c2558 100644 --- a/docs/requirements/Component-SiteCallAudit.md +++ b/docs/requirements/Component-SiteCallAudit.md @@ -148,15 +148,27 @@ and the mailbox supplies the memory barrier between consecutive passes. The daily terminal-row purge uses the same shape. (`NotificationOutboxActor`'s dispatch sweep is the in-repo reference.) -**The central upsert is one statement.** `SiteCallAuditRepository.UpsertAsync` -issues a single batch that runs the monotonic UPDATE first and INSERTs only when -nothing matched *and* the row genuinely does not exist — instead of an -unconditional insert-if-absent followed by the update, which cost two round trips -on every packet and wasted the insert half for every packet after the first. The -existence re-check is load-bearing: a zero row count also means "the monotonic +**The central upsert is one round trip, INSERT-first.** +`SiteCallAuditRepository.UpsertAsync` ships both statements in a single command +text — `IF NOT EXISTS … INSERT;` then the monotonic `UPDATE` — so a packet costs +one round trip, not two. The insert leg is gated on the row's existence *alone*, +never on "the update matched nothing": a zero row count also means "the monotonic guard rejected this packet", and inserting there would fork the mirror with a second row for an id that already exists. +The ORDER is load-bearing, and running the UPDATE first is a data-loss bug. The +two writers — the cached dual-write and the reconciliation pull — routinely carry +*different* lifecycle states for the same `TrackedOperationId`, and both can race +its very first packet. Under UPDATE-first both find no row, so both updates match +nothing; the loser then either fails its existence re-check and skips its insert, +or attempts it and takes a duplicate-key fault — either way its +`Status`/`RetryCount`/`HttpStatus`/`TerminalAtUtc` are dropped, because it never +ran an update against the winner's row. Under INSERT-first the loser's insert is +skipped or faults and its monotonic update still applies to whichever row won, so +the newer state survives every interleaving while a stale one is still rejected by +the rank guard. The duplicate-key catch re-runs the monotonic update for the same +reason (it is idempotent and rank-guarded, so a redundant re-run is inert). + ## Retry / Discard Relay Parked cached calls live in the owning site's S&F buffer. Operator Retry/Discard diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs index 63eab907..1ca8e692 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs @@ -84,6 +84,22 @@ public class AuditLogIngestActor : ReceiveActor /// internal static readonly TimeSpan IngestSqlCommandTimeout = TimeSpan.FromSeconds(15); + /// + /// Budget for the per-row fallback loop, granted as a FRESH + /// rather than reusing the batch's. + /// + /// + /// The fallback used to run on the batch's own token. When the batch failed + /// BECAUSE that token expired, every fallback insert was handed an + /// already-cancelled token and failed instantly: N logged errors, N counter + /// bumps, zero rows accepted, and the site retried the whole packet anyway. + /// A fresh, deliberately SHORT budget (shorter than , + /// so the reply still beats the caller's Ask even when the batch consumed its + /// whole allowance) gives the poison-row isolation the fallback exists for a + /// real chance to land the good rows. + /// + internal static readonly TimeSpan IngestFallbackBudget = TimeSpan.FromSeconds(5); + private readonly IServiceProvider? _serviceProvider; private readonly IAuditLogRepository? _injectedRepository; private readonly ILogger _logger; @@ -265,15 +281,36 @@ public class AuditLogIngestActor : ReceiveActor // back to the per-row path here, so a poison row still costs only // itself. Re-running rows the failed batch may already have committed // is safe: every insert is idempotent on EventId. + // + // The fallback gets its OWN token. Reusing the batch's meant that a + // batch which failed because the budget EXPIRED handed the loop an + // already-cancelled token, so every row failed instantly and the + // fallback did nothing but multiply the log and counter noise by the + // row count. + var budgetExpired = budget.IsCancellationRequested; + using var fallbackBudget = new CancellationTokenSource(IngestFallbackBudget); + _logger.LogWarning(ex, - "Set-based ingest of {Count} audit event(s) failed; falling back to per-row inserts so one bad row does not sink the batch.", - cmd.Events.Count); + "Set-based ingest of {Count} audit event(s) failed{BudgetNote}; falling back to per-row inserts (fresh {FallbackSeconds}s budget) so one bad row does not sink the batch.", + cmd.Events.Count, + budgetExpired ? " after exhausting the ingest budget" : string.Empty, + IngestFallbackBudget.TotalSeconds); + + if (budgetExpired) + { + // A blown budget is ONE failure event for the batch, not one per + // row — bump the health counter once here and suppress the + // per-row bumps below so the dashboard sees a timeout as a + // timeout rather than as N independent write failures. + try { failureCounter?.Increment(); } + catch { /* counter must never throw — defence in depth */ } + } for (var i = 0; i < projected.Count; i++) { try { - await repository.InsertIfNotExistsAsync(projected[i], budget.Token).ConfigureAwait(false); + await repository.InsertIfNotExistsAsync(projected[i], fallbackBudget.Token).ConfigureAwait(false); accepted.Add(cmd.Events[i].EventId); } catch (Exception rowEx) @@ -281,9 +318,14 @@ public class AuditLogIngestActor : ReceiveActor // Per-row catch — one bad row never sinks the whole batch. // The row stays Pending at the site; the next drain retries. // Bump the central health counter so a sustained insert-throw - // failure surfaces on the dashboard. - try { failureCounter?.Increment(); } - catch { /* counter must never throw — defence in depth */ } + // failure surfaces on the dashboard — unless the batch already + // counted itself as a single budget-exhaustion failure. + if (!budgetExpired) + { + try { failureCounter?.Increment(); } + catch { /* counter must never throw — defence in depth */ } + } + _logger.LogError(rowEx, "Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.", cmd.Events[i].EventId); diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/INotificationOutboxRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/INotificationOutboxRepository.cs index 9165e576..b0d40b22 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/INotificationOutboxRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/INotificationOutboxRepository.cs @@ -63,11 +63,25 @@ public interface INotificationOutboxRepository /// payload columns and the dispatcher writes on EVERY attempt. A future /// caller that needs to change an immutable column must add its own method /// rather than widening this one. + /// + /// The return value is the not-found signal. A targeted UPDATE against + /// a row that no longer exists (the retention purge removed it between the + /// read and the write) affects zero rows and is otherwise indistinguishable + /// from success — which is how an operator Retry/Discard came to report + /// success against a vanished notification. Implementations MUST return + /// when no row matched. Callers that speak to a human + /// (the operator one-shots) must surface that as "not found"; the dispatcher + /// treats it as a lost race and logs. + /// /// /// The notification whose delivery state should be persisted. /// Cancellation token. - /// A task that completes when the notification has been persisted. - Task UpdateAsync(Notification n, CancellationToken cancellationToken = default); + /// + /// when the row was found and its delivery state + /// persisted; when no row with that + /// NotificationId exists any more. + /// + Task UpdateAsync(Notification n, CancellationToken cancellationToken = default); /// /// Returns a page of notifications matching , ordered by diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs index a38a6b9c..6bfba4ae 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs @@ -261,14 +261,28 @@ VALUES var occurred = DateTime.SpecifyKind(evt.OccurredAtUtc.UtcDateTime, DateTimeKind.Utc); object actor = string.IsNullOrEmpty(evt.Actor) ? DBNull.Value : evt.Actor; + // NO explicit Size on the string parameters — size 0 means "bind + // the value's own length" (see AddParameter). Declaring the + // COLUMN width here (Actor/Target 256, Action 64, Outcome 16, + // Category 32, SourceNode 64) made SqlClient TRUNCATE an + // over-long value client-side and commit the mutilated row, while + // the single-row path (ExecuteSqlInterpolated, no Size) and the + // reconciliation path sent the same value in full and let the + // server reject it with error 2628. An append-only audit store + // must never silently store a shortened row — there is no + // truncation flag on the record to say it happened — so the batch + // path now defers the length check to the server exactly like + // every other path: reject everywhere, truncate nowhere. + // The explicit SqlDbType stays (it is what fixes the VALUES + // constructor's derived column types, and DateTime2 precision). AddParameter(cmd, "@e" + i, SqlDbType.UniqueIdentifier, size: 0, evt.EventId); AddParameter(cmd, "@t" + i, SqlDbType.DateTime2, size: 0, occurred); - AddParameter(cmd, "@a" + i, SqlDbType.NVarChar, size: 256, actor); - AddParameter(cmd, "@n" + i, SqlDbType.VarChar, size: 64, evt.Action); - AddParameter(cmd, "@o" + i, SqlDbType.VarChar, size: 16, evt.Outcome.ToString()); - AddParameter(cmd, "@c" + i, SqlDbType.VarChar, size: 32, evt.Category); - AddParameter(cmd, "@g" + i, SqlDbType.NVarChar, size: 256, evt.Target); - AddParameter(cmd, "@s" + i, SqlDbType.VarChar, size: 64, evt.SourceNode); + AddParameter(cmd, "@a" + i, SqlDbType.NVarChar, size: 0, actor); + AddParameter(cmd, "@n" + i, SqlDbType.VarChar, size: 0, evt.Action); + AddParameter(cmd, "@o" + i, SqlDbType.VarChar, size: 0, evt.Outcome.ToString()); + AddParameter(cmd, "@c" + i, SqlDbType.VarChar, size: 0, evt.Category); + AddParameter(cmd, "@g" + i, SqlDbType.NVarChar, size: 0, evt.Target); + AddParameter(cmd, "@s" + i, SqlDbType.VarChar, size: 0, evt.SourceNode); AddParameter(cmd, "@r" + i, SqlDbType.UniqueIdentifier, size: 0, evt.CorrelationId); AddParameter(cmd, "@d" + i, SqlDbType.NVarChar, size: -1, evt.DetailsJson); } @@ -290,6 +304,14 @@ VALUES /// which is what makes the VALUES constructor's derived column types stable /// regardless of which rows happen to carry nulls. /// + /// + /// semantics: 0 leaves + /// unset so SqlClient sizes the parameter from the VALUE — the only safe + /// choice for the variable-length audit columns, because an explicit Size + /// SMALLER than the value silently truncates it at bind time instead of + /// letting the server raise 2628. -1 is the explicit MAX marker + /// (nvarchar(max) DetailsJson). Never pass a column width here. + /// private static void AddParameter( System.Data.Common.DbCommand cmd, string name, SqlDbType type, int size, object? value) { diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs index 3d21571e..2232a72a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs @@ -158,7 +158,7 @@ VALUES } /// - public async Task UpdateAsync(Notification n, CancellationToken cancellationToken = default) + public async Task UpdateAsync(Notification n, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(n); @@ -184,7 +184,12 @@ VALUES var notificationId = n.NotificationId; - await _context.Notifications + // The row count is the not-found signal and MUST NOT be discarded. + // DbSet.Update + SaveChanges used to throw DbUpdateConcurrencyException + // when the row had vanished (the daily retention purge deletes terminal + // rows); ExecuteUpdate just reports "0 rows" and returns, which made an + // operator Retry/Discard of a purged notification answer "success". + var rowsAffected = await _context.Notifications .Where(row => row.NotificationId == notificationId) .ExecuteUpdateAsync( setters => setters @@ -196,11 +201,23 @@ VALUES .SetProperty(row => row.NextAttemptAt, nextAttemptAt) .SetProperty(row => row.DeliveredAt, deliveredAt), cancellationToken); + + return rowsAffected > 0; } /// public async Task GetByIdAsync(string notificationId, CancellationToken cancellationToken = default) - => await _context.Notifications.FindAsync(new object[] { notificationId }, cancellationToken); + { + // AsNoTracking + a keyed predicate rather than FindAsync: the write that + // follows this read is ExecuteUpdate, which bypasses the change tracker + // entirely, so tracking the entity bought nothing but a snapshot copy of + // the nvarchar(max) Body/TypeData columns — and left a stale tracked + // instance in the context that a later read of the same id would return + // in preference to the database. FindAsync cannot be made no-tracking. + return await _context.Notifications + .AsNoTracking() + .FirstOrDefaultAsync(row => row.NotificationId == notificationId, cancellationToken); + } /// public async Task<(IReadOnlyList Rows, int TotalCount)> QueryAsync( diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs index ac31c9e0..4d14a7a7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs @@ -1,3 +1,4 @@ +using System.Data; using Microsoft.Data.SqlClient; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; @@ -72,20 +73,26 @@ public class SiteCallAuditRepository : ISiteCallAuditRepository var idText = siteCall.TrackedOperationId.Value.ToString("D"); var incomingRank = GetRankOrThrow(siteCall.Status); - // ONE round trip, UPDATE-first (WP2.2). The predecessor issued an - // unconditional IF NOT EXISTS … INSERT and THEN a monotonic UPDATE — two - // statements, two round trips, on every single packet, of which the insert - // half was wasted work for every packet after the first (the steady state: - // a cached call emits Submitted → Forwarded → Attempted → terminal, so - // three of four packets hit an existing row). + // ONE round trip, INSERT-first. Both statements ship in a single command + // text, so the round-trip saving of the WP2.2 rewrite is preserved — but + // the ORDER is back to insert-then-update, because UPDATE-first LOSES + // DATA under a concurrent first-write. // - // The combined batch below runs UPDATE first and inserts only when the - // UPDATE matched nothing AND the row genuinely does not exist. The - // NOT EXISTS re-check is load-bearing: @@ROWCOUNT = 0 is ALSO what a - // monotonic REJECTION looks like (a stale or regressive packet against an - // existing row), and inserting there would resurrect a row the guard just - // refused. Both statements ship in one command text, so this is one - // round trip, not two. + // The UPDATE-first shape was: UPDATE; SET @updated = @@ROWCOUNT; + // IF @updated = 0 AND NOT EXISTS(…) INSERT. Two writers racing the FIRST + // packet of one TrackedOperationId — the cached dual-write and the + // reconciliation pull, which routinely carry DIFFERENT lifecycle states — + // both find no row, so both UPDATEs match nothing. The loser then either + // fails its own NOT EXISTS re-check (READ COMMITTED, after the winner + // committed) and skips the INSERT, or attempts it and eats a 2627 in the + // catch below. EITHER WAY the loser's Status/RetryCount/HttpStatus/ + // TerminalAtUtc are silently dropped: it never ran an UPDATE against the + // winner's row. + // + // INSERT-first has no such hole. The loser's INSERT is skipped or faults, + // and the monotonic UPDATE that FOLLOWS it applies its state to whichever + // row won — so the newer lifecycle state survives regardless of + // interleaving, and a stale one is still rejected by the rank guard. // // Monotonic update semantics are unchanged: mutate when EITHER the // incoming rank is strictly greater, OR the incoming rank equals the @@ -95,7 +102,19 @@ public class SiteCallAuditRepository : ISiteCallAuditRepository // live instead of freezing at the first Attempted packet. Terminal ranks // are excluded from the tiebreaker, so a later terminal NEVER overwrites // an earlier one; equal stamps are inert (idempotent replay) and a lower - // rank is always a no-op. + // rank is always a no-op. That inertness is what makes the UPDATE + // harmless immediately after this same call's own INSERT: equal rank, + // equal UpdatedAtUtc, zero rows changed. + // + // Raw SQL with explicitly-typed parameters (rather than + // ExecuteSqlInterpolated) so the monotonic UPDATE exists as ONE statement + // text shared by the combined batch and the duplicate-key retry below — + // the predicate is far too intricate to keep in two copies. Explicit + // SqlDbType is load-bearing: an untyped DateTime parameter binds as + // `datetime` (3.33 ms rounding), which would corrupt both the stored + // datetime2 stamps and the `UpdatedAtUtc <` freshness tiebreaker. Sizes + // are deliberately left at the value's own length so the server enforces + // the column widths (see AuditLogRepository.AddParameter). // // SourceNode-stamping: the column is included in the INSERT column list / // VALUES so a fresh row carries the originating node name (node-a/node-b @@ -117,57 +136,11 @@ public class SiteCallAuditRepository : ISiteCallAuditRepository // idempotent. try { - await _context.Database.ExecuteSqlInterpolatedAsync( - $@"DECLARE @updated int; - -UPDATE dbo.SiteCalls -SET Status = {siteCall.Status}, - RetryCount = {siteCall.RetryCount}, - LastError = {siteCall.LastError}, - HttpStatus = {siteCall.HttpStatus}, - UpdatedAtUtc = {siteCall.UpdatedAtUtc}, - TerminalAtUtc = {siteCall.TerminalAtUtc}, - IngestedAtUtc = {siteCall.IngestedAtUtc}, - SourceNode = COALESCE({siteCall.SourceNode}, SourceNode) -WHERE TrackedOperationId = {idText} - AND ( {incomingRank} > (CASE Status - WHEN 'Submitted' THEN 0 - WHEN 'Forwarded' THEN 1 - WHEN 'Attempted' THEN 2 - WHEN 'Skipped' THEN 2 - WHEN 'Delivered' THEN 3 - WHEN 'Failed' THEN 3 - WHEN 'Parked' THEN 3 - WHEN 'Discarded' THEN 3 - ELSE -1 - END) - OR ( {incomingRank} = (CASE Status - WHEN 'Submitted' THEN 0 - WHEN 'Forwarded' THEN 1 - WHEN 'Attempted' THEN 2 - WHEN 'Skipped' THEN 2 - WHEN 'Delivered' THEN 3 - WHEN 'Failed' THEN 3 - WHEN 'Parked' THEN 3 - WHEN 'Discarded' THEN 3 - ELSE -1 - END) - AND {incomingRank} < {TerminalRank} - AND UpdatedAtUtc < {siteCall.UpdatedAtUtc} ) ); - --- Captured IMMEDIATELY after the UPDATE: @@ROWCOUNT is reset by the next --- statement, and reading it inline inside a compound IF condition alongside a --- subquery is not safe (the subquery's own execution can clobber it). -SET @updated = @@ROWCOUNT; - -IF @updated = 0 AND NOT EXISTS (SELECT 1 FROM dbo.SiteCalls WHERE TrackedOperationId = {idText}) -INSERT INTO dbo.SiteCalls - (TrackedOperationId, Channel, Target, SourceSite, SourceNode, Status, RetryCount, - LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, IngestedAtUtc) -VALUES - ({idText}, {siteCall.Channel}, {siteCall.Target}, {siteCall.SourceSite}, {siteCall.SourceNode}, {siteCall.Status}, {siteCall.RetryCount}, - {siteCall.LastError}, {siteCall.HttpStatus}, {siteCall.CreatedAtUtc}, {siteCall.UpdatedAtUtc}, {siteCall.TerminalAtUtc}, {siteCall.IngestedAtUtc});", - ct); + await _context.Database.ExecuteSqlRawAsync( + InsertIfAbsentSql + "\n\n" + MonotonicUpdateSql, + BuildUpsertParameters(siteCall, idText, incomingRank), + ct) + .ConfigureAwait(false); } catch (SqlException ex) when ( ex.Number == SqlErrorUniqueIndexViolation @@ -175,20 +148,112 @@ VALUES { // Two concurrent sessions both found the row absent and both raced to // INSERT; the loser raises 2601/2627 against the TrackedOperationId - // primary key. The winner's row IS the first-write, and this packet's - // content is by construction the same lifecycle state, so the race - // outcome is semantically a no-op. Swallow at Debug — the same - // check-then-act contract the sibling AuditLog/Notification repos - // document. Note the loser's UPDATE leg already ran (against no row), - // so nothing is left half-applied. + // primary key. The winner's row IS the first-write, but it is NOT + // necessarily this packet's lifecycle state — the reconciliation pull + // and the cached dual-write feed this method with different states — + // so the loser MUST still apply its monotonic UPDATE against the + // winner's row. The batch aborted at the faulting INSERT, so re-run + // the UPDATE alone here; it is idempotent and rank-guarded, so + // re-running it is safe even if the batch did reach it. _logger.LogDebug( ex, - "SiteCallAuditRepository.UpsertAsync swallowed duplicate-key violation (error {SqlErrorNumber}) for TrackedOperationId {TrackedOperationId}; treating as no-op.", + "SiteCallAuditRepository.UpsertAsync swallowed duplicate-key violation (error {SqlErrorNumber}) for TrackedOperationId {TrackedOperationId}; re-running the monotonic update against the winning row.", ex.Number, idText); + + await _context.Database.ExecuteSqlRawAsync( + MonotonicUpdateSql, + BuildUpsertParameters(siteCall, idText, incomingRank), + ct) + .ConfigureAwait(false); } } + // Leg 1 of the upsert: create the row when it does not exist yet. Runs FIRST + // so a concurrent first-write loser still has a row to update (see UpsertAsync). + private const string InsertIfAbsentSql = @" +IF NOT EXISTS (SELECT 1 FROM dbo.SiteCalls WHERE TrackedOperationId = @Id) +INSERT INTO dbo.SiteCalls + (TrackedOperationId, Channel, Target, SourceSite, SourceNode, Status, RetryCount, + LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, IngestedAtUtc) +VALUES + (@Id, @Channel, @Target, @SourceSite, @SourceNode, @Status, @RetryCount, + @LastError, @HttpStatus, @CreatedAtUtc, @UpdatedAtUtc, @TerminalAtUtc, @IngestedAtUtc);"; + + // Leg 2 of the upsert: the monotonic guard. Also re-run standalone from the + // duplicate-key catch, which is why it lives in its own constant. + private const string MonotonicUpdateSql = @" +UPDATE dbo.SiteCalls +SET Status = @Status, + RetryCount = @RetryCount, + LastError = @LastError, + HttpStatus = @HttpStatus, + UpdatedAtUtc = @UpdatedAtUtc, + TerminalAtUtc = @TerminalAtUtc, + IngestedAtUtc = @IngestedAtUtc, + SourceNode = COALESCE(@SourceNode, SourceNode) +WHERE TrackedOperationId = @Id + AND ( @Rank > (CASE Status + WHEN 'Submitted' THEN 0 + WHEN 'Forwarded' THEN 1 + WHEN 'Attempted' THEN 2 + WHEN 'Skipped' THEN 2 + WHEN 'Delivered' THEN 3 + WHEN 'Failed' THEN 3 + WHEN 'Parked' THEN 3 + WHEN 'Discarded' THEN 3 + ELSE -1 + END) + OR ( @Rank = (CASE Status + WHEN 'Submitted' THEN 0 + WHEN 'Forwarded' THEN 1 + WHEN 'Attempted' THEN 2 + WHEN 'Skipped' THEN 2 + WHEN 'Delivered' THEN 3 + WHEN 'Failed' THEN 3 + WHEN 'Parked' THEN 3 + WHEN 'Discarded' THEN 3 + ELSE -1 + END) + AND @Rank < @TerminalRank + AND UpdatedAtUtc < @UpdatedAtUtc ) );"; + + /// + /// Builds one FRESH parameter set for the upsert statements. Fresh per call + /// because a instance cannot be attached to two + /// commands, and the duplicate-key retry issues a second command. + /// + private static object[] BuildUpsertParameters(SiteCall siteCall, string idText, int incomingRank) + { + return + [ + Param("@Id", SqlDbType.VarChar, idText), + Param("@Channel", SqlDbType.VarChar, siteCall.Channel), + Param("@Target", SqlDbType.VarChar, siteCall.Target), + Param("@SourceSite", SqlDbType.VarChar, siteCall.SourceSite), + Param("@SourceNode", SqlDbType.VarChar, siteCall.SourceNode), + Param("@Status", SqlDbType.VarChar, siteCall.Status), + Param("@RetryCount", SqlDbType.Int, siteCall.RetryCount), + Param("@LastError", SqlDbType.NVarChar, siteCall.LastError), + Param("@HttpStatus", SqlDbType.Int, siteCall.HttpStatus), + Param("@CreatedAtUtc", SqlDbType.DateTime2, siteCall.CreatedAtUtc), + Param("@UpdatedAtUtc", SqlDbType.DateTime2, siteCall.UpdatedAtUtc), + Param("@TerminalAtUtc", SqlDbType.DateTime2, siteCall.TerminalAtUtc), + Param("@IngestedAtUtc", SqlDbType.DateTime2, siteCall.IngestedAtUtc), + Param("@Rank", SqlDbType.Int, incomingRank), + Param("@TerminalRank", SqlDbType.Int, TerminalRank), + ]; + } + + /// + /// Binds one explicitly-typed parameter, mapping a null CLR value to + /// while KEEPING the declared type. Size is never set — + /// SqlClient sizes from the value, so an over-long string is rejected by the + /// server rather than truncated client-side. + /// + private static SqlParameter Param(string name, SqlDbType type, object? value) => + new(name, type) { Value = value ?? DBNull.Value }; + /// public async Task GetAsync(TrackedOperationId id, CancellationToken ct = default) { diff --git a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs index 9cb04470..19c6e28c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs @@ -603,7 +603,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers notification.Status = NotificationStatus.Parked; notification.LastError = missingAdapterError; notification.LastAttemptAt = now; - await outboxRepository.UpdateAsync(notification, cancellationToken); + await WarnIfVanishedAsync(outboxRepository, notification, cancellationToken); await EmitAttemptAuditAsync( notification, now, @@ -654,7 +654,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers break; } - await outboxRepository.UpdateAsync(notification, cancellationToken); + await WarnIfVanishedAsync(outboxRepository, notification, cancellationToken); // Emit the per-attempt Attempted row exactly once regardless of the // outcome (B2). The error message comes from the outcome, not from @@ -680,6 +680,35 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers } } + /// + /// Persists the dispatcher's delivery-state write and logs when the row has + /// VANISHED underneath it — + /// returning false means the retention purge deleted the notification between + /// the claim and the write. + /// + /// + /// The dispatcher deliberately does not treat this as an error: the delivery + /// itself already happened (or failed) and there is no row left to record the + /// outcome on, so there is nothing to retry or roll back. The audit rows are + /// still emitted — they are the durable record. The operator one-shots + /// (retry/discard) take the opposite stance and answer "not found", because a + /// human is waiting on that answer. + /// + private async Task WarnIfVanishedAsync( + INotificationOutboxRepository repository, + Notification notification, + CancellationToken cancellationToken) + { + var persisted = await repository.UpdateAsync(notification, cancellationToken); + if (!persisted) + { + _logger.LogWarning( + "Notification {NotificationId} disappeared before its delivery state could be written (status {Status}); the row was most likely purged mid-flight.", + notification.NotificationId, + notification.Status); + } + } + /// /// True for , /// , or @@ -1098,7 +1127,16 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers notification.RetryCount = 0; notification.NextAttemptAt = null; notification.LastError = null; - await repository.UpdateAsync(notification); + + // Zero rows updated means the row was purged between the read above and + // this write. Answer the operator honestly instead of reporting a + // re-queue that never happened — and emit no un-park audit row, because + // there is nothing to attribute it to. + if (!await repository.UpdateAsync(notification)) + { + return new RetryNotificationResponse( + request.CorrelationId, Success: false, ErrorMessage: "notification not found"); + } // Operator re-queued a parked notification. Emit a Submitted NotifyDeliver // row attributing the un-park to the operator — otherwise the lifecycle @@ -1170,7 +1208,15 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers } notification.Status = NotificationStatus.Discarded; - await repository.UpdateAsync(notification); + + // Same vanished-row honesty as the retry path: a purge between the read + // and the write leaves nothing to discard, so report not-found rather + // than success (and skip the terminal audit row below). + if (!await repository.UpdateAsync(notification)) + { + return new DiscardNotificationResponse( + request.CorrelationId, Success: false, ErrorMessage: "notification not found"); + } // A manual discard is the OTHER code path that produces // a terminal NotificationStatus transition (alongside the dispatcher). diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs index eff92c94..aee226b5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs @@ -221,7 +221,7 @@ public class SiteCallAuditActor : ReceiveActor ArgumentNullException.ThrowIfNull(repository); ArgumentNullException.ThrowIfNull(logger); - _injectedRepository = repository; + _injectedRepository = new SerializedRepository(repository); _logger = logger; _options = options ?? new SiteCallAuditOptions(); _auditWriter = auditWriter; @@ -267,7 +267,7 @@ public class SiteCallAuditActor : ReceiveActor ArgumentNullException.ThrowIfNull(pullClient); ArgumentNullException.ThrowIfNull(logger); - _injectedRepository = repository; + _injectedRepository = new SerializedRepository(repository); _siteEnumerator = siteEnumerator; _pullClient = pullClient; _logger = logger; @@ -1564,6 +1564,92 @@ public class SiteCallAuditActor : ReceiveActor public static readonly PurgeComplete Instance = new(); private PurgeComplete() { } } + + /// + /// Serializing wrapper applied to a repository handed in through the + /// test constructors. One call at a time, in arrival order. + /// + /// + /// + /// In production every consumer of this actor's repository gets its OWN DI + /// scope — one per message, one per background pass — so the off-mailbox + /// reconciliation and purge passes can safely run alongside mailbox handlers: + /// separate scopes mean separate DbContexts. The injected-repository + /// constructors break that assumption: ONE instance (typically wrapping one + /// DbContext) is shared by the mailbox handlers AND the background + /// passes, and DbContext forbids concurrent operations — "A second + /// operation was started on this context instance" is a hard fault, and the + /// interleaving is nondeterministic, so it surfaces as a flaky test rather + /// than a reliable one. + /// + /// + /// Serializing at the CALL, not around the whole pass, is deliberate: it + /// removes the concurrency hazard while preserving the property the + /// off-mailbox passes exist for — a long drain (blocked in a network pull, + /// holding no repository call) still lets ingest, query and KPI messages be + /// answered. Suspending the mailbox for the duration of a pass would have + /// been simpler and would have invalidated exactly those regression tests. + /// Production is untouched: it never constructs this type. + /// + /// + private sealed class SerializedRepository : ISiteCallAuditRepository + { + private readonly ISiteCallAuditRepository _inner; + private readonly SemaphoreSlim _gate = new(1, 1); + + public SerializedRepository(ISiteCallAuditRepository inner) => _inner = inner; + + public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) => + RunAsync(() => _inner.UpsertAsync(siteCall, ct)); + + public Task GetAsync(TrackedOperationId id, CancellationToken ct = default) => + RunAsync(() => _inner.GetAsync(id, ct)); + + public Task> QueryAsync( + SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) => + RunAsync(() => _inner.QueryAsync(filter, paging, ct)); + + public Task PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) => + RunAsync(() => _inner.PurgeTerminalAsync(olderThanUtc, ct)); + + public Task ComputeKpisAsync( + DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) => + RunAsync(() => _inner.ComputeKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct)); + + public Task> ComputePerSiteKpisAsync( + DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) => + RunAsync(() => _inner.ComputePerSiteKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct)); + + public Task> ComputePerNodeKpisAsync( + DateTime stuckCutoffUtc, DateTime deliveredSinceUtc, CancellationToken ct = default) => + RunAsync(() => _inner.ComputePerNodeKpisAsync(stuckCutoffUtc, deliveredSinceUtc, ct)); + + private async Task RunAsync(Func call) + { + await _gate.WaitAsync().ConfigureAwait(false); + try + { + await call().ConfigureAwait(false); + } + finally + { + _gate.Release(); + } + } + + private async Task RunAsync(Func> call) + { + await _gate.WaitAsync().ConfigureAwait(false); + try + { + return await call().ConfigureAwait(false); + } + finally + { + _gate.Release(); + } + } + } } /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogIngestActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogIngestActorTests.cs index e2d2c927..61fdc239 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogIngestActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogIngestActorTests.cs @@ -322,6 +322,47 @@ public class AuditLogIngestActorTests : TestKit, IClassFixture + /// The per-row fallback must run on its OWN cancellation token, never the + /// batch's. Sharing it meant that when the batch failed BECAUSE the 20 s + /// ingest budget expired, every fallback insert was handed an + /// already-cancelled token: N instant failures, N counter bumps, zero rows + /// accepted — the fallback's entire purpose (land the good rows) defeated at + /// the exact moment it was needed. + /// + /// + /// Pinned by token IDENTITY rather than by expiring a real budget: the budget + /// is a fixed 20 s and a test that waited for it would cost 20 s of wall clock + /// to assert something the token comparison establishes outright. Two + /// s are equal iff they come from the same + /// source, so "not equal" IS "a fresh CTS", and a fresh CTS cannot be + /// pre-cancelled by the batch. + /// + [Fact] + public async Task Receive_WhenBatchFails_PerRowFallbackUsesAFreshToken() + { + var repository = new TokenRecordingRepository(); + var actor = CreateActor(repository); + + var events = Enumerable.Range(0, 3).Select(_ => NewEvent(NewSiteId())).ToList(); + actor.Tell(new IngestAuditEventsCommand(events), TestActor); + + var reply = ExpectMsg(TimeSpan.FromSeconds(10)); + + // The fallback landed every row. + Assert.Equal(3, reply.AcceptedEventIds.Count); + + Assert.NotNull(repository.BatchToken); + Assert.Equal(3, repository.RowTokens.Count); + Assert.All(repository.RowTokens, rowToken => + { + Assert.NotEqual(repository.BatchToken!.Value, rowToken); + Assert.False(rowToken.IsCancellationRequested); + }); + + await Task.CompletedTask; + } + /// Counts how many times the guard's catch surfaced a write failure. private sealed class CountingFailureCounter : ICentralAuditWriteFailureCounter { @@ -329,6 +370,62 @@ public class AuditLogIngestActorTests : TestKit, IClassFixture Count++; } + /// + /// Fails the set-based insert (as an expired budget would) and records the + /// token handed to each write so the fallback's token can be compared with + /// the batch's. + /// + private sealed class TokenRecordingRepository : IAuditLogRepository + { + public CancellationToken? BatchToken { get; private set; } + + public List RowTokens { get; } = new(); + + public Task InsertManyIfNotExistsAsync( + IReadOnlyList events, TimeSpan? commandTimeout = null, CancellationToken ct = default) + { + BatchToken = ct; + throw new OperationCanceledException("simulated ingest-budget expiry", ct); + } + + public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default) + { + RowTokens.Add(ct); + return Task.CompletedTask; + } + + public Task> QueryAsync( + AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task SwitchOutPartitionAsync( + DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task PurgeChannelOlderThanAsync( + string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task BackfillSourceNodeAsync( + string sentinel, DateTime before, int batchSize, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task> GetPartitionBoundariesOlderThanAsync( + DateTime threshold, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task GetKpiSnapshotAsync( + TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task> GetExecutionTreeAsync( + Guid executionId, CancellationToken ct = default) => + throw new NotSupportedException(); + + public Task> GetDistinctSourceNodesAsync(CancellationToken ct = default) => + throw new NotSupportedException(); + } + /// /// Tiny test double that delegates to a real repository but throws on a /// specified EventId. Used to verify per-row failure isolation: one bad diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/AuditLogRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/AuditLogRepositoryTests.cs index 9f880125..fd5b94a1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/AuditLogRepositoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/AuditLogRepositoryTests.cs @@ -707,6 +707,85 @@ public class AuditLogRepositoryTests : IClassFixture Assert.Equal(250, count); } + /// + /// An over-long value must be REJECTED by the server on every write path, not + /// clipped to fit by the client. The set-based path used to declare each + /// string parameter at its COLUMN width (Target/Actor 256, Action 64, + /// Outcome 16, Category 32, SourceNode 64), which makes SqlClient truncate the + /// value at bind time and commit the shortened row — in an APPEND-ONLY audit + /// store, with no PayloadTruncated flag to admit it happened — while the + /// per-row and reconciliation paths sent the same value in full and let the + /// server raise 2628/8152. Reject-everywhere is the contract; this test pins + /// both halves of it. + /// + [SkippableFact] + public async Task InsertManyIfNotExistsAsync_TargetLongerThanColumn_IsRejected_NotTruncated() + { + Skip.IfNot(_fixture.Available, _fixture.SkipReason); + + var siteId = NewSiteId(); + + // dbo.AuditLog.Target is nvarchar(256). + var overlongTarget = new string('T', 300); + var evt = NewEvent( + siteId, + occurredAtUtc: new DateTime(2026, 5, 22, 12, 0, 0, DateTimeKind.Utc), + target: overlongTarget); + + await using var batchContext = CreateContext(); + var batchRepo = new AuditLogRepository(batchContext); + + // 2628 = "String or binary data would be truncated in table …" (SQL 2019+), + // 8152 = its pre-2019 predecessor. Either proves the server saw the full value. + var batchEx = await Assert.ThrowsAsync( + () => batchRepo.InsertManyIfNotExistsAsync(new[] { evt })); + Assert.Contains(batchEx.Number, new[] { 2628, 8152 }); + + // The per-row path — which the batch path falls back to — rejects it too. + await using var rowContext = CreateContext(); + var rowRepo = new AuditLogRepository(rowContext); + var rowEx = await Assert.ThrowsAsync( + () => rowRepo.InsertIfNotExistsAsync(evt)); + Assert.Contains(rowEx.Number, new[] { 2628, 8152 }); + + // Nothing landed — in particular, no 256-character mutilated copy. + await using var readContext = CreateContext(); + var rows = await readContext.Set() + .Where(e => e.SourceSiteId == siteId) + .ToListAsync(); + + Assert.Empty(rows); + } + + /// + /// The companion to the rejection test: a value that FITS must still round-trip + /// through the set-based path byte for byte, at the exact column boundary. + /// + [SkippableFact] + public async Task InsertManyIfNotExistsAsync_TargetAtColumnLimit_RoundTripsIntact() + { + Skip.IfNot(_fixture.Available, _fixture.SkipReason); + + var siteId = NewSiteId(); + var boundaryTarget = new string('T', 256); + var evt = NewEvent( + siteId, + occurredAtUtc: new DateTime(2026, 5, 22, 12, 30, 0, DateTimeKind.Utc), + target: boundaryTarget); + + await using var context = CreateContext(); + var repo = new AuditLogRepository(context); + Assert.Equal(1, await repo.InsertManyIfNotExistsAsync(new[] { evt })); + + await using var readContext = CreateContext(); + var loaded = await readContext.Set() + .Where(e => e.SourceSiteId == siteId) + .ToListAsync(); + + Assert.Single(loaded); + Assert.Equal(boundaryTarget, loaded[0].Target); + } + [SkippableFact] public async Task InsertManyIfNotExistsAsync_EmptyBatch_IsANoOp() { @@ -1346,12 +1425,14 @@ public class AuditLogRepositoryTests : IClassFixture string? errorMessage = null, Guid? executionId = null, Guid? parentExecutionId = null, - string? sourceNode = null) => + string? sourceNode = null, + string? target = null) => ScadaBridgeAuditEventFactory.Create( channel: channel, kind: kind, status: status, occurredAtUtc: occurredAtUtc, + target: target, sourceNode: sourceNode, sourceSiteId: siteId, executionId: executionId, diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/SiteCallAuditRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/SiteCallAuditRepositoryTests.cs index 9244e8ae..f1353bb6 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/SiteCallAuditRepositoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/SiteCallAuditRepositoryTests.cs @@ -101,15 +101,12 @@ public class SiteCallAuditRepositoryTests : IClassFixture { Skip.IfNot(_fixture.Available, _fixture.SkipReason); - // WP2.2 collapsed the two-round-trip "insert-if-absent then monotonic - // update" into ONE batch that updates first and inserts only when nothing - // matched. That makes @@ROWCOUNT = 0 ambiguous: it means "no such row" - // AND "the monotonic guard rejected this packet". Guarding the insert on - // @@ROWCOUNT alone would therefore let every stale/regressive packet - // append a SECOND row for an id that already exists — silently forking - // the mirror. The re-check for the row's existence is what prevents that, - // and this pins it: after a rejected regressive upsert there is still - // exactly ONE row, still carrying the advanced state. + // A rejected packet must leave the mirror untouched — one row, still + // carrying the advanced state. The hazard this pins is the insert leg + // firing for a row that already exists and forking the mirror into two + // rows for one id: the insert is gated on IF NOT EXISTS alone (never on + // "the update matched nothing", which is ALSO what a monotonic rejection + // looks like), so a stale or regressive packet is a pure no-op. var id = TrackedOperationId.New(); await using var context = CreateContext(); var repo = new SiteCallAuditRepository(context); @@ -137,6 +134,109 @@ public class SiteCallAuditRepositoryTests : IClassFixture Assert.Null(loaded[0].LastError); } + /// + /// The deterministic half of the concurrent-first-write regression: the OTHER + /// writer already committed the row, so this call's insert leg is skipped + /// entirely and only the monotonic UPDATE can carry its state in. Under the + /// UPDATE-first shape the loser of a first-write race ran its UPDATE BEFORE + /// any row existed and then skipped its INSERT, dropping the packet's + /// Status/RetryCount/HttpStatus/TerminalAtUtc on the floor. + /// + [SkippableFact] + public async Task UpsertAsync_RowAlreadyCreatedByAnotherWriter_StillAppliesThisPacketsState() + { + Skip.IfNot(_fixture.Available, _fixture.SkipReason); + + var id = TrackedOperationId.New(); + var t0 = new DateTime(2026, 5, 22, 8, 0, 0, DateTimeKind.Utc); + + // Writer 1 (say the reconciliation pull) wins the insert with an early state. + await using var winnerContext = CreateContext(); + await new SiteCallAuditRepository(winnerContext).UpsertAsync( + NewRow(id, status: "Submitted", createdAtUtc: t0, updatedAtUtc: t0)); + + // Writer 2 (the cached dual-write) carries a LATER lifecycle state for the + // same id, on its own connection. + await using var loserContext = CreateContext(); + var loserRepo = new SiteCallAuditRepository(loserContext); + await loserRepo.UpsertAsync(NewRow( + id, + status: "Parked", + retryCount: 4, + lastError: "gave up", + httpStatus: 503, + createdAtUtc: t0, + updatedAtUtc: t0.AddSeconds(30), + terminal: true)); + + await using var readContext = CreateContext(); + var loaded = await readContext.Set() + .Where(s => s.TrackedOperationId == id) + .ToListAsync(); + + Assert.Single(loaded); + Assert.Equal("Parked", loaded[0].Status); + Assert.Equal(4, loaded[0].RetryCount); + Assert.Equal(503, loaded[0].HttpStatus); + Assert.Equal("gave up", loaded[0].LastError); + Assert.NotNull(loaded[0].TerminalAtUtc); + } + + /// + /// The real-concurrency half: two writers racing the FIRST packet of the same + /// id, on separate connections, carrying DIFFERENT lifecycle states. However + /// the interleaving falls — either insert winning, or the loser eating a 2627 + /// on its insert leg — the NEWER state must be the one on the row afterwards, + /// and exactly one row must exist. + /// + /// + /// Repeated over several ids because the interleaving is not controllable; one + /// id would make the test a coin flip on whether it visits the racing path at + /// all. Every iteration must hold regardless of which way it raced, so the + /// test itself is not flaky — only its coverage of the rare branch is + /// probabilistic. + /// + [SkippableFact] + public async Task UpsertAsync_TwoConcurrentFirstWrites_NewerStateSurvives() + { + Skip.IfNot(_fixture.Available, _fixture.SkipReason); + + var t0 = new DateTime(2026, 5, 22, 9, 0, 0, DateTimeKind.Utc); + + for (var i = 0; i < 12; i++) + { + var id = TrackedOperationId.New(); + + await using var contextA = CreateContext(); + await using var contextB = CreateContext(); + var repoA = new SiteCallAuditRepository(contextA); + var repoB = new SiteCallAuditRepository(contextB); + + var early = NewRow(id, status: "Submitted", createdAtUtc: t0, updatedAtUtc: t0); + var late = NewRow( + id, + status: "Delivered", + retryCount: 2, + createdAtUtc: t0, + updatedAtUtc: t0.AddSeconds(10), + terminal: true); + + await Task.WhenAll( + Task.Run(() => repoA.UpsertAsync(early)), + Task.Run(() => repoB.UpsertAsync(late))); + + await using var readContext = CreateContext(); + var loaded = await readContext.Set() + .Where(s => s.TrackedOperationId == id) + .ToListAsync(); + + Assert.Single(loaded); + Assert.Equal("Delivered", loaded[0].Status); + Assert.Equal(2, loaded[0].RetryCount); + Assert.NotNull(loaded[0].TerminalAtUtc); + } + } + [SkippableFact] public async Task UpsertAsync_SameStatus_EqualUpdatedAt_IsNoOp() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs index f24ecdeb..e4e486d7 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs @@ -537,7 +537,7 @@ public class NotificationOutboxRepositoryTests : IDisposable var loaded = await _repository.GetByIdAsync(id); loaded!.Status = NotificationStatus.Delivered; loaded.DeliveredAt = new DateTimeOffset(2026, 5, 19, 9, 0, 0, TimeSpan.Zero); - await _repository.UpdateAsync(loaded); + Assert.True(await _repository.UpdateAsync(loaded)); _context.ChangeTracker.Clear(); var reloaded = await _context.Notifications.FindAsync(id); @@ -545,6 +545,33 @@ public class NotificationOutboxRepositoryTests : IDisposable Assert.Equal(new DateTimeOffset(2026, 5, 19, 9, 0, 0, TimeSpan.Zero), reloaded.DeliveredAt); } + /// + /// The write is a targeted UPDATE that bypasses the change tracker, so a row + /// deleted between the read and the write simply matches nothing. The row + /// count is the ONLY not-found signal available, and callers (the operator + /// retry/discard one-shots) depend on it to avoid reporting success against a + /// purged notification. + /// + [Fact] + public async Task UpdateAsync_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); + + // The daily retention purge removes the row while the operator's request + // is in flight. + await _context.Notifications.Where(n => n.NotificationId == id).ExecuteDeleteAsync(); + _context.ChangeTracker.Clear(); + + loaded!.Status = NotificationStatus.Pending; + Assert.False(await _repository.UpdateAsync(loaded)); + } + [Fact] public async Task QueryAsync_AppliesFilters_OrdersByCreatedAtDescending_AndPaginates() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Ingest/NotificationIngestTypeStampingTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Ingest/NotificationIngestTypeStampingTests.cs index 36eefa58..b09624d0 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Ingest/NotificationIngestTypeStampingTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Ingest/NotificationIngestTypeStampingTests.cs @@ -21,7 +21,7 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests.Ingest; public class NotificationIngestTypeStampingTests : TestKit { private readonly INotificationOutboxRepository _outboxRepository = - Substitute.For(); + OutboxRepositorySubstitute.Healthy(); private readonly INotificationRepository _listRepository = Substitute.For(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorAttemptEmissionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorAttemptEmissionTests.cs index ccef7308..f5bbeb1b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorAttemptEmissionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorAttemptEmissionTests.cs @@ -26,7 +26,7 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; public class NotificationOutboxActorAttemptEmissionTests : TestKit { private readonly INotificationOutboxRepository _outboxRepository = - Substitute.For(); + OutboxRepositorySubstitute.Healthy(); private readonly INotificationRepository _notificationRepository = Substitute.For(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorAuditInjectionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorAuditInjectionTests.cs index dce94903..be39c1f3 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorAuditInjectionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorAuditInjectionTests.cs @@ -21,7 +21,7 @@ public class NotificationOutboxActorAuditInjectionTests : TestKit private static IServiceProvider BuildEmptyProvider() { var services = new ServiceCollection(); - services.AddScoped(_ => Substitute.For()); + services.AddScoped(_ => OutboxRepositorySubstitute.Healthy()); services.AddScoped(_ => Substitute.For()); return services.BuildServiceProvider(); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs index dc093a93..eb00e790 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs @@ -22,7 +22,7 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; public class NotificationOutboxActorDispatchTests : TestKit { private readonly INotificationOutboxRepository _outboxRepository = - Substitute.For(); + OutboxRepositorySubstitute.Healthy(); private readonly INotificationRepository _notificationRepository = Substitute.For(); @@ -491,7 +491,7 @@ public class NotificationOutboxActorDispatchTests : TestKit // INotificationOutboxRepository registration with a private counting factory, // so we don't mutate the shared _outboxRepository field that other tests in // this class configure differently. - var outboxRepository = Substitute.For(); + var outboxRepository = OutboxRepositorySubstitute.Healthy(); // De-race (S11): hand out a fresh due notification for the FIRST THREE claims, then // an empty batch forever. This caps the deliverable work — and therefore the // UpdateAsync count — at exactly three, no matter how many dispatch ticks the diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorIngestTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorIngestTests.cs index 5b387b20..6eb6b160 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorIngestTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorIngestTests.cs @@ -21,7 +21,7 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; public class NotificationOutboxActorIngestTests : TestKit { private readonly INotificationOutboxRepository _repository = - Substitute.For(); + OutboxRepositorySubstitute.Healthy(); private IServiceProvider BuildServiceProvider() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorPurgeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorPurgeTests.cs index bb8c0964..c397958a 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorPurgeTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorPurgeTests.cs @@ -21,7 +21,7 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; public class NotificationOutboxActorPurgeTests : TestKit { private readonly INotificationOutboxRepository _outboxRepository = - Substitute.For(); + OutboxRepositorySubstitute.Healthy(); private readonly INotificationRepository _notificationRepository = Substitute.For(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorQueryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorQueryTests.cs index a0fc75de..5e93a010 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorQueryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorQueryTests.cs @@ -22,7 +22,7 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; public class NotificationOutboxActorQueryTests : TestKit { private readonly INotificationOutboxRepository _repository = - Substitute.For(); + OutboxRepositorySubstitute.Healthy(); private IServiceProvider BuildServiceProvider() { @@ -298,6 +298,48 @@ public class NotificationOutboxActorQueryTests : TestKit Assert.Contains("not found", response.ErrorMessage); } + /// + /// 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 + /// told the notification is gone — reporting a re-queue that never happened + /// is what the ExecuteUpdate rewrite silently introduced (the predecessor + /// threw DbUpdateConcurrencyException here). + /// + [Fact] + public void Retry_RowPurgedBeforeWrite_RepliesNotFound() + { + var row = MakeNotification(status: NotificationStatus.Parked, retryCount: 10, lastError: "gave up"); + _repository.GetByIdAsync(row.NotificationId, Arg.Any()).Returns(row); + _repository.UpdateAsync(Arg.Any(), Arg.Any()).Returns(false); + var actor = CreateActor(); + + actor.Tell(new RetryNotificationRequest("corr-vanished-retry", row.NotificationId), TestActor); + + var response = ExpectMsg(); + Assert.Equal("corr-vanished-retry", response.CorrelationId); + Assert.False(response.Success); + Assert.NotNull(response.ErrorMessage); + Assert.Contains("not found", response.ErrorMessage); + } + + /// Discard half of . + [Fact] + public void Discard_RowPurgedBeforeWrite_RepliesNotFound() + { + var row = MakeNotification(status: NotificationStatus.Parked); + _repository.GetByIdAsync(row.NotificationId, Arg.Any()).Returns(row); + _repository.UpdateAsync(Arg.Any(), Arg.Any()).Returns(false); + var actor = CreateActor(); + + actor.Tell(new DiscardNotificationRequest("corr-vanished-discard", row.NotificationId), TestActor); + + var response = ExpectMsg(); + Assert.Equal("corr-vanished-discard", response.CorrelationId); + Assert.False(response.Success); + Assert.NotNull(response.ErrorMessage); + Assert.Contains("not found", response.ErrorMessage); + } + [Fact] public void Discard_ParkedNotification_MarksDiscarded_AndSucceeds() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorRetryEmissionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorRetryEmissionTests.cs index 804b939f..df07b46a 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorRetryEmissionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorRetryEmissionTests.cs @@ -25,7 +25,7 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; public class NotificationOutboxActorRetryEmissionTests : TestKit { private readonly INotificationOutboxRepository _outboxRepository = - Substitute.For(); + OutboxRepositorySubstitute.Healthy(); private readonly RecordingCentralAuditWriter _auditWriter = new(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorTerminalEmissionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorTerminalEmissionTests.cs index c8afb4a8..887f075b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorTerminalEmissionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorTerminalEmissionTests.cs @@ -27,7 +27,7 @@ namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; public class NotificationOutboxActorTerminalEmissionTests : TestKit { private readonly INotificationOutboxRepository _outboxRepository = - Substitute.For(); + OutboxRepositorySubstitute.Healthy(); private readonly INotificationRepository _notificationRepository = Substitute.For(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/OutboxRepositorySubstitute.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/OutboxRepositorySubstitute.cs new file mode 100644 index 00000000..67f9e444 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/OutboxRepositorySubstitute.cs @@ -0,0 +1,31 @@ +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Notifications; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; + +namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; + +/// +/// Factory for the outbox repository substitute used across these tests. +/// +/// +/// returns whether the +/// targeted UPDATE actually matched a row — false is the "this +/// notification no longer exists" signal the operator retry/discard handlers +/// turn into a not-found reply. NSubstitute's default for Task<bool> +/// is false, i.e. the VANISHED-row answer, so an unconfigured substitute +/// would silently put every test on the failure path. Tests that want a healthy +/// store take one from here; the vanished-row tests configure +/// Returns(false) for themselves. +/// +internal static class OutboxRepositorySubstitute +{ + /// Creates a substitute whose writes report that the row was found. + public static INotificationOutboxRepository Healthy() + { + var repository = Substitute.For(); + repository + .UpdateAsync(Arg.Any(), Arg.Any()) + .Returns(true); + return repository; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs index d64d3c8c..e131c40a 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs @@ -674,6 +674,164 @@ public class SiteCallAuditReconciliationTests : TestKit client.Release(); } + /// + /// An injected repository is ONE instance — typically wrapping one + /// DbContext — shared by the mailbox handlers and the off-mailbox + /// reconciliation/purge passes, and DbContext forbids concurrent + /// operations. Every call the actor makes through an injected repository must + /// therefore be serialized, so a drain's upserts never overlap an ingest + /// upsert arriving on the mailbox. (Production is unaffected: each message and + /// each pass resolves its own scope, hence its own context.) + /// + [Fact] + public async Task InjectedRepository_IsNeverCalledConcurrently_ByDrainAndMailbox() + { + var siteId = "siteBusy"; + var sites = new StaticEnumerator(new SiteEntry(siteId, "http://siteBusy:8083")); + + // One page of rows for the drain to upsert, each call held open long + // enough that a concurrent mailbox upsert would land inside it. + var pulled = Enumerable.Range(0, 6) + .Select(_ => NewRow(TrackedOperationId.New(), siteId)) + .ToArray(); + var client = new OneBatchThenEmptyPullClient(pulled); + var repo = new ConcurrencyDetectingRepo(TimeSpan.FromMilliseconds(40)); + + var actor = CreateActor(sites, client, repo, FastTickOptions()); + + // Once the pull has been served the drain is upserting; flood the mailbox + // with ingest commands so the two writers overlap in wall-clock time. + await client.Entered.WaitAsync(TimeSpan.FromSeconds(5)); + + var asks = Enumerable.Range(0, 6) + .Select(_ => actor.Ask( + new UpsertSiteCallCommand(NewRow(TrackedOperationId.New(), siteId)), + TimeSpan.FromSeconds(10))) + .ToArray(); + + await Task.WhenAll(asks); + Assert.All(asks, ask => Assert.True(ask.Result.Accepted)); + + // The drain's own upserts must have run in the same window. + AwaitAssert( + () => Assert.True(repo.UpsertCount >= 12, $"expected both writers to have run; saw {repo.UpsertCount}"), + TimeSpan.FromSeconds(5)); + + Assert.Equal(1, repo.MaxObservedConcurrency); + } + + /// + /// Serves one page of rows on the first pull and nothing afterwards, so the + /// drain has real upsert work to do and then settles. + /// + private sealed class OneBatchThenEmptyPullClient : IPullSiteCallsClient + { + private readonly SiteCall[] _rows; + private readonly TaskCompletionSource _entered = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _callCount; + + public OneBatchThenEmptyPullClient(SiteCall[] rows) => _rows = rows; + + public Task Entered => _entered.Task; + + public Task PullAsync( + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) + { + var first = Interlocked.Increment(ref _callCount) == 1; + _entered.TrySetResult(); + return Task.FromResult(new PullSiteCallsResponse( + first ? _rows : Array.Empty(), MoreAvailable: false)); + } + } + + /// + /// Records the peak number of overlapping repository calls. Each call is held + /// open briefly so an overlap, if the actor allows one, is observed rather + /// than missed by timing luck. + /// + private sealed class ConcurrencyDetectingRepo : ISiteCallAuditRepository + { + private readonly TimeSpan _hold; + private int _inFlight; + private int _maxObserved; + private int _upsertCount; + + public ConcurrencyDetectingRepo(TimeSpan hold) => _hold = hold; + + public int MaxObservedConcurrency => Volatile.Read(ref _maxObserved); + + public int UpsertCount => Volatile.Read(ref _upsertCount); + + public async Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) + { + Interlocked.Increment(ref _upsertCount); + await TrackAsync().ConfigureAwait(false); + } + + public async Task GetAsync(TrackedOperationId id, CancellationToken ct = default) + { + await TrackAsync().ConfigureAwait(false); + return null; + } + + public async Task> QueryAsync( + SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) + { + await TrackAsync().ConfigureAwait(false); + return Array.Empty(); + } + + public async Task PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) + { + await TrackAsync().ConfigureAwait(false); + return 0; + } + + public async Task ComputeKpisAsync( + DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) + { + await TrackAsync().ConfigureAwait(false); + return new SiteCallKpiSnapshot(0, 0, 0, 0, null, 0); + } + + public async Task> ComputePerSiteKpisAsync( + DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) + { + await TrackAsync().ConfigureAwait(false); + return Array.Empty(); + } + + public async Task> ComputePerNodeKpisAsync( + DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) + { + await TrackAsync().ConfigureAwait(false); + return Array.Empty(); + } + + private async Task TrackAsync() + { + var current = Interlocked.Increment(ref _inFlight); + + // Monotonic max without a lock. + int seen; + while (current > (seen = Volatile.Read(ref _maxObserved)) + && Interlocked.CompareExchange(ref _maxObserved, current, seen) != seen) + { + // Another thread moved the max; re-read and retry. + } + + try + { + await Task.Delay(_hold).ConfigureAwait(false); + } + finally + { + Interlocked.Decrement(ref _inFlight); + } + } + } + /// /// that also counts invocations, so a test /// can prove no second pass started while the first was blocked.