fix: outbox KPI per-status legs, composite event-log indexes, AuditLog ingest FORCESEEK (#122, #123, #124)

#122: the Notification Outbox KPI snapshots ORed every status in one aggregation, which scanned the clustered index once the table held ~1 M Parked rows (0.7-2.6 s; Health dashboard up to 5 s). Each snapshot now runs three status-equality legs that seek their own index (~0.12 s); a cold Health load on the rig is 0.12-0.52 s. #123: site_events event_type/severity indexes are now composites with timestamp, so a filtered page is read in order and stops at the limit (severity=Error over 1 GB: 2,287 -> 15 ms cold, 115 -> 1.6 ms warm). Costs 454 -> 530 B/row, ~14 % fewer rows under the same cap. The schema apply drops the old indexes in place; the cap size model samples timestamp three times and re-scores inside its band with unchanged constants. #124: the set-based AuditLog insert's NOT EXISTS probe scanned the EventId index and concurrent writers deadlocked on S/IX page locks. WITH (FORCESEEK): 4 writers 775 rows/s + 76 deadlocks in 30 s -> 5,214 rows/s + 0. New concurrent-writer regression test fails 4/4 without the hint.
2026-11-10 16:46:42 -05:00
parent 2b80a6defa
commit 424db1555d
12 changed files with 386 additions and 158 deletions
@@ -8,9 +8,11 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests;

 /// <summary>
 /// Query-shape coverage for the Notification Outbox KPI snapshots (arch-review 04).
-/// Each snapshot must be a single-query conditional aggregation with a
-/// bounded oldest-pending — the old shape issued 5-7 sequential COUNT round trips
-/// and materialized the whole live queue to find the oldest row.
+/// Each snapshot is at most three status-equality legs (live queue, parked, delivered
+/// in the interval) — the shape before arch-review 04 issued 5-7 sequential COUNT round
+/// trips and materialized the whole live queue to find the oldest row, and the single
+/// OR'd aggregation that replaced it scanned the clustered index once the table held
+/// ~1 M Parked rows (#122).
 /// </summary>
 public class NotificationOutboxRepositoryKpiQueryShapeTests
 {
@@ -25,10 +27,10 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
            SourceNode = sourceNode,
        };

-    /// <summary>Verifies that all three KPI snapshots issue at most two queries and bound the oldest-pending lookup.</summary>
+    /// <summary>Verifies that all three KPI snapshots issue at most three seekable legs and never OR the parked and live predicates.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
-    public async Task AllThreeKpiSnapshots_IssueAtMostTwoQueries_AndBoundTheOldestLookup()
+    public async Task AllThreeKpiSnapshots_IssueAtMostThreeStatusLegs_NeverOneOrdScan()
    {
        var counter = new SelectCountingInterceptor();
        await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter);
@@ -45,13 +47,12 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
        var stuckCutoff = now.AddMinutes(-5);
        var deliveredSince = now.AddMinutes(-60);

-        // Global: one aggregation query + one bounded oldest query.
+        // Global: three legs — live (counts + MIN), parked count, delivered count.
        counter.Reset();
        var global = await repo.ComputeKpisAsync(stuckCutoff, deliveredSince);
-        Assert.True(counter.Count <= 2,
-            $"ComputeKpisAsync issued {counter.Count} queries against Notifications; expected <= 2");
-        // The oldest lookup must be bounded (LIMIT/TOP), never a full non-terminal SELECT.
-        Assert.Contains(counter.Commands, sql => sql.Contains("LIMIT", StringComparison.OrdinalIgnoreCase));
+        Assert.True(counter.Count <= 3,
+            $"ComputeKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
+        AssertNoLegMixesParkedWithLive(counter.Commands);
        // The aggregation must be PREDICATE-RESTRICTED, not an unrestricted
        // scan. Every KPI here concerns the live queue, the parked backlog or the
        // last delivery interval; historical Delivered/Discarded rows — which are
@@ -68,8 +69,9 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
        // Per-site: a single grouped conditional-aggregation query (counts + MIN).
        counter.Reset();
        var perSite = await repo.ComputePerSiteKpisAsync(stuckCutoff, deliveredSince);
-        Assert.True(counter.Count <= 2,
-            $"ComputePerSiteKpisAsync issued {counter.Count} queries against Notifications; expected <= 2");
+        Assert.True(counter.Count <= 3,
+            $"ComputePerSiteKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
+        AssertNoLegMixesParkedWithLive(counter.Commands);
        var siteA = Assert.Single(perSite);
        Assert.Equal(3, siteA.QueueDepth);
        Assert.NotNull(siteA.OldestPendingAge);
@@ -77,13 +79,49 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
        // Per-node: same single grouped aggregation, keyed by SourceNode.
        counter.Reset();
        var perNode = await repo.ComputePerNodeKpisAsync(stuckCutoff, deliveredSince);
-        Assert.True(counter.Count <= 2,
-            $"ComputePerNodeKpisAsync issued {counter.Count} queries against Notifications; expected <= 2");
+        Assert.True(counter.Count <= 3,
+            $"ComputePerNodeKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
+        AssertNoLegMixesParkedWithLive(counter.Commands);
        var nodeA = Assert.Single(perNode);
        Assert.Equal(3, nodeA.QueueDepth);
        Assert.NotNull(nodeA.OldestPendingAge);
    }

+    /// <summary>Verifies the three legs merge per key and a key with only one leg still surfaces.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task PerSiteKpis_MergeLegsPerKey()
+    {
+        await using var ctx = SqliteTestHelper.CreateInMemoryContext(new SelectCountingInterceptor());
+        var now = DateTimeOffset.UtcNow;
+        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Pending, now.AddMinutes(-30)));
+        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Parked, now.AddMinutes(-30)));
+        ctx.Notifications.Add(NewNotification("plant-b", "node-b", NotificationStatus.Parked, now.AddMinutes(-30)));
+        ctx.Notifications.Add(NewNotification("plant-c", "node-a", NotificationStatus.Delivered, now.AddMinutes(-30), now.AddMinutes(-1)));
+        ctx.Notifications.Add(NewNotification("plant-d", "node-b", NotificationStatus.Discarded, now.AddMinutes(-30)));
+        await ctx.SaveChangesAsync();
+
+        var repo = new NotificationOutboxRepository(ctx);
+        var sites = await repo.ComputePerSiteKpisAsync(now.AddMinutes(-5), now.AddMinutes(-60));
+
+        Assert.Equal(new[] { "plant-a", "plant-b", "plant-c" }, sites.Select(s => s.SourceSiteId));
+        Assert.Equal((1, 1, 1, 0), (sites[0].QueueDepth, sites[0].StuckCount, sites[0].ParkedCount, sites[0].DeliveredLastInterval));
+        Assert.NotNull(sites[0].OldestPendingAge);
+        Assert.Equal((0, 1, 0), (sites[1].QueueDepth, sites[1].ParkedCount, sites[1].DeliveredLastInterval));
+        Assert.Null(sites[1].OldestPendingAge);
+        Assert.Equal((0, 0, 1), (sites[2].QueueDepth, sites[2].ParkedCount, sites[2].DeliveredLastInterval));
+
+        var global = await repo.ComputeKpisAsync(now.AddMinutes(-5), now.AddMinutes(-60));
+        Assert.Equal((1, 1, 2, 1), (global.QueueDepth, global.StuckCount, global.ParkedCount, global.DeliveredLastInterval));
+    }
+
+    // An OR across the parked and live predicates is what made SQL Server scan the
+    // clustered index on a large table (#122); each leg must name one status set.
+    private static void AssertNoLegMixesParkedWithLive(IEnumerable<string> commands) =>
+        Assert.All(commands, sql => Assert.False(
+            sql.Contains("'Parked'", StringComparison.Ordinal) && sql.Contains("'Pending'", StringComparison.Ordinal),
+            $"a KPI leg ORs Parked with the live queue:\n{sql}"));
+
    /// <summary>
    /// Records the text of every reader command touching the Notifications table so a
    /// test can count the round trips a KPI snapshot makes and inspect their shape.
@@ -635,6 +635,65 @@ public class AuditLogRepositoryTests : IClassFixture<MsSqlMigrationFixture>
            rows.Select(r => r.EventId).OrderBy(g => g).ToArray());
    }

+    /// <summary>
+    /// Verifies that concurrent batch inserts do not deadlock each other (#124). Before the
+    /// probe was forced to seek, four writers of 256-row packets lost most packets to error
+    /// 1205 within seconds: the anti-semi-join scanned the EventId index and the writers
+    /// deadlocked on each other's S/IX page locks.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [SkippableFact]
+    public async Task InsertManyIfNotExistsAsync_ConcurrentWriters_DoNotDeadlock()
+    {
+        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
+
+        const int writers = 4, packetsPerWriter = 15, packetSize = 256, seedPackets = 40;
+        var siteId = NewSiteId();
+        var baseTime = new DateTime(2026, 5, 22, 8, 0, 0, DateTimeKind.Utc);
+
+        // Every packet spans the SAME time window, as live telemetry from several sites
+        // does, so the writers' rows interleave on the same clustered pages. The seed
+        // grows the index first: the deadlock rate rose with table size (bench, 4
+        // writers: 3 victims in 65 packets on an empty table, 17 in 52 at 5 k rows).
+        List<IReadOnlyList<AuditEvent>> Packets(int count, int offset) =>
+            Enumerable.Range(offset, count)
+                .Select(p => (IReadOnlyList<AuditEvent>)Enumerable.Range(0, packetSize)
+                    .Select(i => NewEvent(siteId, occurredAtUtc: baseTime.AddMilliseconds(i).AddTicks(p)))
+                    .ToList())
+                .ToList();
+
+        foreach (var seed in Packets(seedPackets, 0))
+        {
+            await using var context = CreateContext();
+            await new AuditLogRepository(context).InsertManyIfNotExistsAsync(seed);
+        }
+
+        var packets = Packets(writers * packetsPerWriter, seedPackets);
+
+        var deadlocks = 0;
+        await Task.WhenAll(Enumerable.Range(0, writers).Select(w => Task.Run(async () =>
+        {
+            for (var p = w; p < packets.Count; p += writers)
+            {
+                try
+                {
+                    await using var context = CreateContext();
+                    await new AuditLogRepository(context).InsertManyIfNotExistsAsync(packets[p]);
+                }
+                catch (SqlException ex) when (ex.Number == 1205)
+                {
+                    Interlocked.Increment(ref deadlocks);
+                }
+            }
+        })));
+
+        Assert.Equal(0, deadlocks);
+
+        await using var readContext = CreateContext();
+        var count = await readContext.Set<AuditLogRow>().CountAsync(e => e.SourceSiteId == siteId);
+        Assert.Equal((seedPackets + (writers * packetsPerWriter)) * packetSize, count);
+    }
+
    /// <summary>Verifies that duplicate event ids within one batch collapse to a single row.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [SkippableFact]