perf(central): set-based ingest, aligned partition purge, KPI query shapes, EF hygiene

src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs

@@ -18,8 +18,9 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
 /// ingest timestamp into DetailsJson (there is no promoted IngestedAtUtc
 /// column — the value is a DetailsJson field set via
 /// <see cref="AuditRowProjection.WithIngestedAtUtc"/>) and inserted idempotently
-/// via <see cref="IAuditLogRepository.InsertIfNotExistsAsync"/> — duplicates are
-/// silently swallowed (first-write-wins).
+/// via <see cref="IAuditLogRepository.InsertManyIfNotExistsAsync"/> — duplicates
+/// are silently swallowed (first-write-wins), whether they repeat inside one
+/// packet or across packets.
 /// </summary>
 /// <remarks>
 /// <para>
@@ -28,9 +29,11 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
 /// consistent and the site is free to flip its local row to <c>Forwarded</c>.
 /// </para>
 /// <para>
-/// Audit-write failures must NEVER abort the user-facing action. The actor
-/// wraps each repository call in its own try/catch so a single bad row cannot
-/// cause the rest of the batch to be lost, and it guards scope/repository
+/// Audit-write failures must NEVER abort the user-facing action. Each message
+/// is written as ONE set-based statement (or one transaction, for the cached
+/// dual-write); if that fails the actor retries the same work row-by-row inside
+/// per-row try/catch, so a single bad row still cannot cause the rest of the
+/// batch to be lost. It also guards scope/repository
 /// resolution so a transient DI fault cannot restart the singleton — those
 /// catches are what keep this actor alive across handler throws, not the
 /// supervisor strategy. The <see cref="SupervisorStrategy"/> override returns
@@ -50,6 +53,31 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
 /// </remarks>
 public class AuditLogIngestActor : ReceiveActor
 {
+    /// <summary>
+    /// Overall budget for one ingest message's database work, deliberately
+    /// SHORTER than the gRPC Ask that wraps it.
+    /// </summary>
+    /// <remarks>
+    /// The path used to stack three identical 30 s budgets — the site's Ask
+    /// (<c>CommunicationOptions.NotificationForwardTimeout</c>), the central gRPC
+    /// handler's Ask (<c>SiteStreamGrpcServer.AuditIngestAskTimeout</c>) and the
+    /// ADO.NET command default — so they all expired at the same instant. The
+    /// caller therefore learned nothing except "it took 30 s": no partial ack, no
+    /// distinction between a slow database and a wedged singleton. Making the
+    /// innermost budget strictly smallest means a slow batch is abandoned by the
+    /// actor FIRST, with the accepted-so-far ids still replied, while the outer
+    /// Asks are still waiting.
+    /// </remarks>
+    internal static readonly TimeSpan IngestBudget = TimeSpan.FromSeconds(20);
+
+    /// <summary>
+    /// Per-statement SQL timeout for the ingest write, strictly inside
+    /// <see cref="IngestBudget"/> so a single wedged statement surfaces as a SQL
+    /// timeout the per-batch catch can log, rather than consuming the whole
+    /// actor-level budget and starving the rest of the batch.
+    /// </summary>
+    internal static readonly TimeSpan IngestSqlCommandTimeout = TimeSpan.FromSeconds(15);
+
     private readonly IServiceProvider? _serviceProvider;
     private readonly IAuditLogRepository? _injectedRepository;
     private readonly ILogger<AuditLogIngestActor> _logger;
@@ -185,39 +213,75 @@ public class AuditLogIngestActor : ReceiveActor
         DateTime nowUtc,
         List<Guid> accepted)
     {
+        // Stamp the ingest timestamp here, not at the site. Redact BEFORE the
+        // IngestedAtUtc stamp so the redacted copy carries the central-side
+        // ingest timestamp. The redactor is contract-bound to never throw; a null
+        // redactor (test composition root, no IAuditRedactor registered) falls
+        // back to the SafeDefault rather than pass-through, so HTTP header
+        // redaction always runs. IngestedAtUtc is a DetailsJson field on the
+        // canonical record, so stamp it via the projection helper.
+        var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
+        var projected = new List<AuditEvent>(cmd.Events.Count);
         foreach (var evt in cmd.Events)
         {
-            try
+            var filtered = safeRedactor.Apply(evt);
+            projected.Add(AuditRowProjection.WithIngestedAtUtc(filtered, nowUtc));
+        }
+
+        // ONE set-based statement for the whole packet (WP2.2) instead of one
+        // IF NOT EXISTS … INSERT round trip per event. A telemetry packet of 200
+        // rows used to cost 200 sequential round trips on the central singleton's
+        // dispatcher; it now costs two. Idempotency is unchanged — duplicates
+        // within the packet collapse first-write-wins, duplicates against
+        // committed rows are eliminated by the anti-semi-join, and a concurrent
+        // writer degrades to the repository's per-row fallback.
+        using var budget = new CancellationTokenSource(IngestBudget);
+        try
+        {
+            await repository
+                .InsertManyIfNotExistsAsync(projected, IngestSqlCommandTimeout, budget.Token)
+                .ConfigureAwait(false);
+
+            // A batch that returned without throwing means every row is now
+            // present — inserted here or already committed by an earlier delivery.
+            // Both count as accepted: the storage state is consistent and the site
+            // is free to flip its local rows to Forwarded.
+            foreach (var evt in cmd.Events)
             {
-                // Stamp the ingest timestamp here, not at the site. The
-                // repository's duplicate-key hardening already swallows
-                // duplicate-key races, so the same id arriving twice (site
-                // retry, reconciliation) is a silent no-op.
-                // Redact BEFORE the IngestedAtUtc stamp so the redacted
-                // copy carries the central-side ingest timestamp. The redactor
-                // is contract-bound to never throw. A null
-                // redactor (test composition root, no IAuditRedactor
-                // registered) now falls back to the SafeDefault rather than
-                // pass-through, so HTTP header redaction always runs.
-                // IngestedAtUtc is a DetailsJson field on
-                // the canonical record, so stamp it via the projection helper.
-                var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
-                var filtered = safeRedactor.Apply(evt);
-                var ingested = AuditRowProjection.WithIngestedAtUtc(filtered, nowUtc);
-                await repository.InsertIfNotExistsAsync(ingested).ConfigureAwait(false);
                 accepted.Add(evt.EventId);
             }
-            catch (Exception ex)
+        }
+        catch (Exception ex)
+        {
+            // The batch is a throughput optimisation, NOT a change to the
+            // failure grain. The documented invariant — "a single bad row cannot
+            // cause the rest of the batch to be lost" — is preserved by falling
+            // 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.
+            _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);
+
+            for (var i = 0; i < projected.Count; i++)
             {
-                // 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 */ }
-                _logger.LogError(ex,
-                    "Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.",
-                    evt.EventId);
+                try
+                {
+                    await repository.InsertIfNotExistsAsync(projected[i], budget.Token).ConfigureAwait(false);
+                    accepted.Add(cmd.Events[i].EventId);
+                }
+                catch (Exception rowEx)
+                {
+                    // 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 */ }
+                    _logger.LogError(rowEx,
+                        "Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.",
+                        cmd.Events[i].EventId);
+                }
             }
         }
     }
@@ -262,6 +326,24 @@ public class AuditLogIngestActor : ReceiveActor
 
             var strategy = dbContext.Database.CreateExecutionStrategy();
 
+            // Fast path (WP2.2): the WHOLE packet in ONE transaction — one
+            // set-based audit insert plus one single-statement upsert per entry.
+            // The predecessor opened a transaction PER entry and issued three
+            // statements inside it (IF NOT EXISTS insert, then the two-statement
+            // SiteCalls upsert), so a 50-entry packet cost ~200 round trips and 50
+            // commits. If anything faults, the per-entry loop below re-runs the
+            // packet with its original per-entry isolation, so the documented
+            // "one entry's failure does not abort the others" invariant survives —
+            // it is simply no longer paid for on the healthy path.
+            using var budget = new CancellationTokenSource(IngestBudget);
+            if (await TryIngestCachedBatchAsync(
+                    strategy, dbContext, auditRepo, siteCallRepo, redactor, cmd, accepted, budget.Token)
+                .ConfigureAwait(false))
+            {
+                replyTo.Tell(new IngestCachedTelemetryReply(accepted));
+                return;
+            }
+
             foreach (var entry in cmd.Entries)
             {
                 try
@@ -330,6 +412,90 @@ public class AuditLogIngestActor : ReceiveActor
         replyTo.Tell(new IngestCachedTelemetryReply(accepted));
     }
 
+    /// <summary>
+    /// Attempts the whole cached-telemetry packet as ONE transaction: a single
+    /// set-based audit insert followed by one monotonic <c>SiteCalls</c> upsert
+    /// per entry. Returns <see langword="true"/> when it committed (and only then
+    /// appends to <paramref name="accepted"/>), <see langword="false"/> when the
+    /// caller should fall back to the per-entry transaction loop.
+    /// </summary>
+    /// <remarks>
+    /// The batch is all-or-nothing by construction — it is a single transaction —
+    /// which is why a failure MUST fall back rather than report partial success:
+    /// the per-entry loop is what turns one poison entry into one lost entry
+    /// instead of a lost packet. Nothing is appended to
+    /// <paramref name="accepted"/> until the commit returns, so a failed attempt
+    /// leaves the caller's list untouched and the retry starts from a clean slate.
+    /// </remarks>
+    private async Task<bool> TryIngestCachedBatchAsync(
+        Microsoft.EntityFrameworkCore.Storage.IExecutionStrategy strategy,
+        ScadaBridgeDbContext dbContext,
+        IAuditLogRepository auditRepo,
+        ISiteCallAuditRepository siteCallRepo,
+        IAuditRedactor? redactor,
+        IngestCachedTelemetryCommand cmd,
+        List<Guid> accepted,
+        CancellationToken ct)
+    {
+        try
+        {
+            await strategy.ExecuteAsync(async () =>
+            {
+                await using var tx = await dbContext.Database
+                    .BeginTransactionAsync(ct)
+                    .ConfigureAwait(false);
+
+                // One central-side instant for the whole packet so a join across
+                // the two tables sees matching timestamps (debugging convenience,
+                // not a correctness invariant).
+                var ingestedAt = DateTime.UtcNow;
+                var safeRedactor = redactor ?? SafeDefaultAuditRedactor.Instance;
+
+                var auditRows = new List<AuditEvent>(cmd.Entries.Count);
+                foreach (var entry in cmd.Entries)
+                {
+                    // Only the AuditLog row's payload columns are redactable;
+                    // SiteCalls carries operational state only (status, retry
+                    // count) and is left untouched.
+                    var filteredAudit = safeRedactor.Apply(entry.Audit);
+                    auditRows.Add(AuditRowProjection.WithIngestedAtUtc(filteredAudit, ingestedAt));
+                }
+
+                await auditRepo
+                    .InsertManyIfNotExistsAsync(auditRows, IngestSqlCommandTimeout, ct)
+                    .ConfigureAwait(false);
+
+                foreach (var entry in cmd.Entries)
+                {
+                    await siteCallRepo
+                        .UpsertAsync(entry.SiteCall with { IngestedAtUtc = ingestedAt }, ct)
+                        .ConfigureAwait(false);
+                }
+
+                await tx.CommitAsync(ct).ConfigureAwait(false);
+            }).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            // No health-counter bump here — the per-entry retry is the authority
+            // on whether this packet genuinely failed, and double-counting a
+            // batch that the fallback then writes successfully would make the
+            // dashboard read as a sustained fault during normal contention.
+            _logger.LogWarning(
+                ex,
+                "Batched cached-telemetry dual-write of {Count} entr(ies) failed; falling back to per-entry transactions.",
+                cmd.Entries.Count);
+            return false;
+        }
+
+        foreach (var entry in cmd.Entries)
+        {
+            accepted.Add(entry.Audit.EventId);
+        }
+
+        return true;
+    }
+
     /// <summary>
     /// Fallback handler installed on the single-repository test ctor — that
     /// ctor has no DbContext and no <see cref="ISiteCallAuditRepository"/>, so