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.
This commit is contained in:
Joseph Doherty
2026-11-10 16:46:42 -05:00
parent 2b80a6defa
commit 424db1555d
12 changed files with 386 additions and 158 deletions
+3 -2
View File
File diff suppressed because one or more lines are too long
+6 -2
View File
@@ -57,12 +57,16 @@ cmd.CommandText = """
details TEXT details TEXT
); );
CREATE INDEX IF NOT EXISTS idx_events_timestamp ON site_events(timestamp); CREATE INDEX IF NOT EXISTS idx_events_timestamp ON site_events(timestamp);
CREATE INDEX IF NOT EXISTS idx_events_type ON site_events(event_type);
CREATE INDEX IF NOT EXISTS idx_events_instance ON site_events(instance_id); CREATE INDEX IF NOT EXISTS idx_events_instance ON site_events(instance_id);
CREATE INDEX IF NOT EXISTS idx_events_severity ON site_events(severity); DROP INDEX IF EXISTS idx_events_type;
DROP INDEX IF EXISTS idx_events_severity;
CREATE INDEX IF NOT EXISTS idx_events_type_timestamp ON site_events(event_type, timestamp);
CREATE INDEX IF NOT EXISTS idx_events_severity_timestamp ON site_events(severity, timestamp);
"""; """;
``` ```
The `event_type` and `severity` indexes are composites with `timestamp` (#123). Every query orders by `(timestamp, id)` and returns one page, so with a single-column index an equality filter read and sorted every matching row: `severity = Error` over a full 1 GB log took 2.3 s cold and 115 ms warm. The composite index returns rows already in timestamp order and stops at the page limit, in about 1 ms. The cost is about 69 bytes more index per row, which the storage-cap model charges, so a 1 GB cap now holds roughly 15 % fewer events. The two `DROP INDEX` statements migrate a pre-#123 file in place on its first boot; rebuilding the two indexes on a full 1 GB table takes seconds and happens once. Indexes are not part of LocalDb's replicated-table digest, so this change does not stop a mixed-version pair from syncing.
Keyword search (`KeywordFilter`) runs as `LIKE '%…%' ESCAPE '\'` on `message` and `source`. A leading-wildcard `LIKE` cannot use a B-tree index, so keyword-only queries full-scan the table. All other filters (`event_type`, `severity`, `instance_id`, `timestamp`) are covered by the indexes above. Keyword search (`KeywordFilter`) runs as `LIKE '%…%' ESCAPE '\'` on `message` and `source`. A leading-wildcard `LIKE` cannot use a B-tree index, so keyword-only queries full-scan the table. All other filters (`event_type`, `severity`, `instance_id`, `timestamp`) are covered by the indexes above.
### Connection lock ### Connection lock
@@ -171,6 +171,24 @@ stream whose row width changes sharply *in the middle* of the history — neithe
tail — is mis-sized until the trim reaches it. Self-correcting in practice, since the cap is tail — is mis-sized until the trim reaches it. Self-correcting in practice, since the cap is
re-evaluated every tick and the head sample advances as the oldest rows go. re-evaluated every tick and the head sample advances as the oldest rows go.
### 3.2 Recalibration for the composite indexes (#123, 2026-11-10)
#123 replaced the single-column `event_type` and `severity` indexes with `(event_type, timestamp)`
and `(severity, timestamp)`. There are still four secondary indexes, so `IndexedColumnCount = 4`
stands. The sampler's `indexed_bytes` now counts `timestamp` three times: once for its own index
and once for each composite. The other constants are unchanged.
The model was re-scored with `EstimateSiteEventsSize_TracksTheRealFootprint`:
| Workload | estimated | actual | error | before #123 |
| --- | ---: | ---: | ---: | ---: |
| narrow rows, no details | 3 859 626 | 3 551 232 | **+8.7 %** | +8.4 % |
| typical rows, 200 B details | 4 520 096 | 4 005 888 | **+12.8 %** | +12.8 % |
| wide rows, 2 KB details | 13 012 048 | 13 078 528 | **−0.5 %** | −0.8 % |
The band and the high bias are unchanged. `SiteEventsSchemaCalibrationTests` pins the new index
set and the columns each index covers.
## 4. The wipe-prevention guard, restated ## 4. The wipe-prevention guard, restated
The register asked whether the "size did not decrease" guard should survive. It should, and it The register asked whether the "size did not decrease" guard should survive. It should, and it
@@ -250,7 +250,18 @@ VALUES
} }
sql.Append("\n) AS v (").Append(CanonicalColumnList).Append(")\n"); sql.Append("\n) AS v (").Append(CanonicalColumnList).Append(")\n");
sql.Append("WHERE NOT EXISTS (SELECT 1 FROM dbo.AuditLog x WHERE x.EventId = v.EventId);"); // FORCESEEK (#124): left to itself the optimizer answered the anti-semi-join for
// a 256-row packet by SCANNING the EventId index. Under READ COMMITTED a scan
// holds its current page's S lock while it requests the next page, so two
// concurrent inserts — each holding IX on the pages it had just written —
// deadlocked on each other's pages (error 1205; captured graph: two INSERTs,
// S/IX page locks on AuditLog). Measured with 4 concurrent writers on a fresh
// database (ResidualBench audit-contention): 76 victims in 30 s, 775 rows/s.
// One index seek per VALUES row locks only the key it reads, and only while
// reading it: 0 victims, 5,214 rows/s. A seek can still wait on another
// writer's uncommitted row with the SAME EventId — a genuine duplicate, which
// the duplicate-key fallback below already resolves.
sql.Append("WHERE NOT EXISTS (SELECT 1 FROM dbo.AuditLog x WITH (FORCESEEK) WHERE x.EventId = v.EventId);");
var conn = _context.Database.GetDbConnection(); var conn = _context.Database.GetDbConnection();
var openedHere = false; var openedHere = false;
@@ -428,63 +428,50 @@ VALUES
return total; return total;
} }
// ── Outbox KPIs (#122) ────────────────────────────────────────────────────────────
//
// Every snapshot is computed as THREE status-equality legs instead of one
// aggregation over an OR'd predicate:
// * live — Status IN (Pending, Retrying): queue depth, stuck count, oldest age;
// * parked — Status = Parked: the parked count;
// * delivered — Status = Delivered AND DeliveredAt >= since: the interval count.
// The OR'd pre-filter (the shape before #122) made SQL Server scan the clustered
// index: on the rig's 4.35 M-row table (945 k Parked, 3.4 M Delivered) the global
// snapshot took 0.7-2.6 s and 76 k logical reads. Split, each leg seeks its own index
// — (Status, NextAttemptAt) for live and parked, the filtered IX_Notifications_Delivered
// for the interval leg — and the global snapshot takes ~0.12 s (parked count 18 k
// reads, live 195, delivered 5). The grouped per-site/per-node snapshots still read
// every parked row to group it, but the live and delivered legs no longer ride along.
/// <inheritdoc /> /// <inheritdoc />
public async Task<NotificationKpiSnapshot> ComputeKpisAsync( public async Task<NotificationKpiSnapshot> ComputeKpisAsync(
DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince, CancellationToken cancellationToken = default) DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince, CancellationToken cancellationToken = default)
{ {
var now = DateTimeOffset.UtcNow; var now = DateTimeOffset.UtcNow;
// One conditional-aggregation pass replaces four sequential COUNT round trips: // Live leg: counts and the oldest CreatedAt in one aggregation over the (small)
// each metric is a COUNT(CASE WHEN <predicate> THEN 1 END) over the same scan // non-terminal set. GroupBy(_ => 1) yields a single group, or none when the queue is empty.
// (arch-review 04). GroupBy(_ => 1) yields a single group (no rows → no group). var live = await LiveQueue()
//
// The aggregation is PREDICATE-RESTRICTED instead of scanning the
// whole table. Every KPI here is about the live queue (Pending/Retrying), the
// parked backlog, or the last delivery interval; the overwhelming bulk of the
// Notifications table is historical Delivered and Discarded rows that
// contribute to NONE of them. The pre-filter below is the union of the
// metric-contributing predicates, which lets the optimizer seek the status
// index for the live/parked legs and the filtered
// (DeliveredAt) WHERE Status='Delivered' index for the interval leg,
// instead of paying a full scan whose cost grows with retained history.
// Mirrors the pre-filter ComputePerSiteKpisAsync/ComputePerNodeKpisAsync
// already use — the global snapshot was the odd one out.
var counts = await _context.Notifications
.Where(n => n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying
|| n.Status == NotificationStatus.Parked
|| (n.Status == NotificationStatus.Delivered
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince))
.GroupBy(_ => 1) .GroupBy(_ => 1)
.Select(g => new .Select(g => new
{ {
QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending QueueDepth = g.Count(),
|| n.Status == NotificationStatus.Retrying), StuckCount = g.Count(n => n.CreatedAt < stuckCutoff),
StuckCount = g.Count(n => (n.Status == NotificationStatus.Pending OldestCreatedAt = g.Min(n => (DateTimeOffset?)n.CreatedAt),
|| n.Status == NotificationStatus.Retrying)
&& n.CreatedAt < stuckCutoff),
ParkedCount = g.Count(n => n.Status == NotificationStatus.Parked),
DeliveredLastInterval = g.Count(n => n.Status == NotificationStatus.Delivered
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince),
}) })
.FirstOrDefaultAsync(cancellationToken); .FirstOrDefaultAsync(cancellationToken);
// Oldest non-terminal CreatedAt — TOP(1) via ordered FirstOrDefault, so the var parkedCount = await _context.Notifications
// live queue is never materialized. The nullable cast lets FirstOrDefault .CountAsync(n => n.Status == NotificationStatus.Parked, cancellationToken);
// return null (no non-terminal rows) instead of throwing.
var oldestCreatedAt = await _context.Notifications var deliveredCount = await DeliveredSince(deliveredSince).CountAsync(cancellationToken);
.Where(n => n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying)
.OrderBy(n => n.CreatedAt)
.Select(n => (DateTimeOffset?)n.CreatedAt)
.FirstOrDefaultAsync(cancellationToken);
return new NotificationKpiSnapshot( return new NotificationKpiSnapshot(
QueueDepth: counts?.QueueDepth ?? 0, QueueDepth: live?.QueueDepth ?? 0,
StuckCount: counts?.StuckCount ?? 0, StuckCount: live?.StuckCount ?? 0,
ParkedCount: counts?.ParkedCount ?? 0, ParkedCount: parkedCount,
DeliveredLastInterval: counts?.DeliveredLastInterval ?? 0, DeliveredLastInterval: deliveredCount,
OldestPendingAge: oldestCreatedAt is null ? null : now - oldestCreatedAt.Value); OldestPendingAge: live?.OldestCreatedAt is null ? null : now - live.OldestCreatedAt.Value);
} }
/// <inheritdoc /> /// <inheritdoc />
@@ -493,47 +480,31 @@ VALUES
{ {
var now = DateTimeOffset.UtcNow; var now = DateTimeOffset.UtcNow;
// One grouped conditional-aggregation query per scope: counts AND the oldest var live = await LiveQueue()
// non-terminal CreatedAt (server-side MIN over a CASE, so the live queue is
// never materialized) computed in a single scan. Pre-filter to rows that
// contribute to at least one metric so a site with only fully-uncounted rows
// (e.g. Discarded, or out-of-window Delivered) does not surface as an
// all-zero row — preserving the prior union-of-metric-keys semantics.
var rows = await _context.Notifications
.Where(n => n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying
|| n.Status == NotificationStatus.Parked
|| (n.Status == NotificationStatus.Delivered
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince))
.GroupBy(n => n.SourceSiteId) .GroupBy(n => n.SourceSiteId)
.Select(g => new .Select(g => new LiveRow(
{ g.Key,
Site = g.Key, g.Count(),
QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending g.Count(n => n.CreatedAt < stuckCutoff),
|| n.Status == NotificationStatus.Retrying), g.Min(n => (DateTimeOffset?)n.CreatedAt)))
StuckCount = g.Count(n => (n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying)
&& n.CreatedAt < stuckCutoff),
ParkedCount = g.Count(n => n.Status == NotificationStatus.Parked),
DeliveredLastInterval = g.Count(n => n.Status == NotificationStatus.Delivered
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince),
OldestCreatedAt = g.Min(n => (n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying)
? (DateTimeOffset?)n.CreatedAt
: null),
})
.ToListAsync(cancellationToken); .ToListAsync(cancellationToken);
return rows var parked = await _context.Notifications
.OrderBy(r => r.Site, StringComparer.Ordinal) .Where(n => n.Status == NotificationStatus.Parked)
.Select(r => new SiteNotificationKpiSnapshot( .GroupBy(n => n.SourceSiteId)
SourceSiteId: r.Site, .Select(g => new CountRow(g.Key, g.Count()))
QueueDepth: r.QueueDepth, .ToListAsync(cancellationToken);
StuckCount: r.StuckCount,
ParkedCount: r.ParkedCount, var delivered = await DeliveredSince(deliveredSince)
DeliveredLastInterval: r.DeliveredLastInterval, .GroupBy(n => n.SourceSiteId)
OldestPendingAge: r.OldestCreatedAt is null ? null : now - r.OldestCreatedAt.Value)) .Select(g => new CountRow(g.Key, g.Count()))
.ToList(); .ToListAsync(cancellationToken);
// A key appears only when it contributes to at least one leg, so a site with only
// uncounted rows (Discarded, out-of-window Delivered) never surfaces as an all-zero row.
return MergeLegs(live, parked, delivered, now,
(key, depth, stuck, parkedCount, deliveredCount, age) =>
new SiteNotificationKpiSnapshot(key, depth, stuck, parkedCount, deliveredCount, age));
} }
/// <inheritdoc /> /// <inheritdoc />
@@ -542,44 +513,83 @@ VALUES
{ {
var now = DateTimeOffset.UtcNow; var now = DateTimeOffset.UtcNow;
// Same single-scan grouped aggregation as ComputePerSiteKpisAsync, keyed by // Same three legs as ComputePerSiteKpisAsync, keyed by SourceNode. Rows with a NULL
// SourceNode. Rows with a NULL SourceNode (legacy / unstamped) are excluded — // SourceNode (legacy / unstamped) are excluded — per-node KPIs are only meaningful
// per-node KPIs are only meaningful when node identity is known. // when node identity is known.
var rows = await _context.Notifications var live = await LiveQueue()
.Where(n => n.SourceNode != null .Where(n => n.SourceNode != null)
&& (n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying
|| n.Status == NotificationStatus.Parked
|| (n.Status == NotificationStatus.Delivered
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince)))
.GroupBy(n => n.SourceNode!) .GroupBy(n => n.SourceNode!)
.Select(g => new .Select(g => new LiveRow(
{ g.Key,
Node = g.Key, g.Count(),
QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending g.Count(n => n.CreatedAt < stuckCutoff),
|| n.Status == NotificationStatus.Retrying), g.Min(n => (DateTimeOffset?)n.CreatedAt)))
StuckCount = g.Count(n => (n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying)
&& n.CreatedAt < stuckCutoff),
ParkedCount = g.Count(n => n.Status == NotificationStatus.Parked),
DeliveredLastInterval = g.Count(n => n.Status == NotificationStatus.Delivered
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince),
OldestCreatedAt = g.Min(n => (n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying)
? (DateTimeOffset?)n.CreatedAt
: null),
})
.ToListAsync(cancellationToken); .ToListAsync(cancellationToken);
return rows var parked = await _context.Notifications
.OrderBy(r => r.Node, StringComparer.Ordinal) .Where(n => n.Status == NotificationStatus.Parked && n.SourceNode != null)
.Select(r => new NodeNotificationKpiSnapshot( .GroupBy(n => n.SourceNode!)
SourceNode: r.Node, .Select(g => new CountRow(g.Key, g.Count()))
QueueDepth: r.QueueDepth, .ToListAsync(cancellationToken);
StuckCount: r.StuckCount,
ParkedCount: r.ParkedCount, var delivered = await DeliveredSince(deliveredSince)
DeliveredLastInterval: r.DeliveredLastInterval, .Where(n => n.SourceNode != null)
OldestPendingAge: r.OldestCreatedAt is null ? null : now - r.OldestCreatedAt.Value)) .GroupBy(n => n.SourceNode!)
.Select(g => new CountRow(g.Key, g.Count()))
.ToListAsync(cancellationToken);
return MergeLegs(live, parked, delivered, now,
(key, depth, stuck, parkedCount, deliveredCount, age) =>
new NodeNotificationKpiSnapshot(key, depth, stuck, parkedCount, deliveredCount, age));
}
/// <summary>The non-terminal queue: <c>Status IN (Pending, Retrying)</c>, two index seek ranges.</summary>
private IQueryable<Notification> LiveQueue() =>
_context.Notifications.Where(n => n.Status == NotificationStatus.Pending
|| n.Status == NotificationStatus.Retrying);
/// <summary>
/// Delivered rows inside the KPI interval. The literal <c>Status = 'Delivered'</c> plus the
/// <c>DeliveredAt</c> range matches the filtered <c>IX_Notifications_Delivered</c> index.
/// </summary>
private IQueryable<Notification> DeliveredSince(DateTimeOffset deliveredSince) =>
_context.Notifications.Where(n => n.Status == NotificationStatus.Delivered
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince);
private sealed record LiveRow(string Key, int QueueDepth, int StuckCount, DateTimeOffset? OldestCreatedAt);
private sealed record CountRow(string Key, int Count);
/// <summary>
/// Joins the three per-key legs into one snapshot per key, ordered ordinally by key.
/// </summary>
private static List<TSnapshot> MergeLegs<TSnapshot>(
IReadOnlyList<LiveRow> live,
IReadOnlyList<CountRow> parked,
IReadOnlyList<CountRow> delivered,
DateTimeOffset now,
Func<string, int, int, int, int, TimeSpan?, TSnapshot> create)
{
var liveByKey = live.ToDictionary(r => r.Key, StringComparer.Ordinal);
var parkedByKey = parked.ToDictionary(r => r.Key, r => r.Count, StringComparer.Ordinal);
var deliveredByKey = delivered.ToDictionary(r => r.Key, r => r.Count, StringComparer.Ordinal);
return liveByKey.Keys
.Concat(parkedByKey.Keys)
.Concat(deliveredByKey.Keys)
.Distinct(StringComparer.Ordinal)
.OrderBy(k => k, StringComparer.Ordinal)
.Select(key =>
{
liveByKey.TryGetValue(key, out var l);
return create(
key,
l?.QueueDepth ?? 0,
l?.StuckCount ?? 0,
parkedByKey.GetValueOrDefault(key),
deliveredByKey.GetValueOrDefault(key),
l?.OldestCreatedAt is null ? null : now - l.OldestCreatedAt.Value);
})
.ToList(); .ToList();
} }
@@ -89,7 +89,11 @@ public class EventLogPurgeService : BackgroundService
/// </summary> /// </summary>
private const double TablePageFillFactor = 0.78; private const double TablePageFillFactor = 0.78;
/// <summary>Number of indexes on <c>site_events</c> (timestamp, event_type, instance_id, severity).</summary> /// <summary>
/// Number of secondary indexes on <c>site_events</c>: timestamp, instance_id,
/// (event_type, timestamp) and (severity, timestamp). The sampled <c>indexed_bytes</c>
/// charges every covered column once per index, hence <c>3 * octet_length(timestamp)</c>.
/// </summary>
private const int IndexedColumnCount = 4; private const int IndexedColumnCount = 4;
/// <summary>Per-entry bookkeeping in an index B-tree leaf: key-length varint, rowid and cell pointer.</summary> /// <summary>Per-entry bookkeeping in an index B-tree leaf: key-length varint, rowid and cell pointer.</summary>
@@ -680,8 +684,8 @@ public class EventLogPurgeService : BackgroundService
/// <para> /// <para>
/// <b>The model.</b> Rows are sampled from both ends of the table (<see cref="SizeSampleRows"/> /// <b>The model.</b> Rows are sampled from both ends of the table (<see cref="SizeSampleRows"/>
/// from each — the head is what the oldest-first trim will actually delete, the tail catches a /// from each — the head is what the oldest-first trim will actually delete, the tail catches a
/// workload whose event shape has drifted) to get an average payload width and an average /// workload whose event shape has drifted) to get an average payload width and the average
/// width across the four indexed columns. Table cost is then computed as whole pages — /// summed width of the columns each of the four indexes covers. Table cost is then computed as whole pages —
/// <c>ceil(rows / rowsPerPage) × pageSize</c> — rather than as a flat bytes-per-row rate, /// <c>ceil(rows / rowsPerPage) × pageSize</c> — rather than as a flat bytes-per-row rate,
/// because that is the term that dominates once rows approach half a page: rows too wide to /// because that is the term that dominates once rows approach half a page: rows too wide to
/// pack two-per-page waste the remainder, which a linear model underestimates by ~40 %. /// pack two-per-page waste the remainder, which a linear model underestimates by ~40 %.
@@ -730,7 +734,7 @@ public class EventLogPurgeService : BackgroundService
+ octet_length(severity) + COALESCE(octet_length(instance_id), 0) + octet_length(severity) + COALESCE(octet_length(instance_id), 0)
+ octet_length(source) + octet_length(message) + octet_length(source) + octet_length(message)
+ COALESCE(octet_length(details), 0) AS payload_bytes, + COALESCE(octet_length(details), 0) AS payload_bytes,
octet_length(timestamp) + octet_length(event_type) (3 * octet_length(timestamp)) + octet_length(event_type)
+ octet_length(severity) + COALESCE(octet_length(instance_id), 0) + octet_length(severity) + COALESCE(octet_length(instance_id), 0)
AS indexed_bytes AS indexed_bytes
FROM site_events LIMIT {SizeSampleRows} FROM site_events LIMIT {SizeSampleRows}
@@ -742,7 +746,7 @@ public class EventLogPurgeService : BackgroundService
+ octet_length(severity) + COALESCE(octet_length(instance_id), 0) + octet_length(severity) + COALESCE(octet_length(instance_id), 0)
+ octet_length(source) + octet_length(message) + octet_length(source) + octet_length(message)
+ COALESCE(octet_length(details), 0) AS payload_bytes, + COALESCE(octet_length(details), 0) AS payload_bytes,
octet_length(timestamp) + octet_length(event_type) (3 * octet_length(timestamp)) + octet_length(event_type)
+ octet_length(severity) + COALESCE(octet_length(instance_id), 0) + octet_length(severity) + COALESCE(octet_length(instance_id), 0)
AS indexed_bytes AS indexed_bytes
FROM site_events ORDER BY rowid DESC LIMIT {SizeSampleRows} FROM site_events ORDER BY rowid DESC LIMIT {SizeSampleRows}
@@ -67,10 +67,24 @@ public static class SiteEventLogSchema
details TEXT details TEXT
); );
CREATE INDEX IF NOT EXISTS idx_events_timestamp ON site_events(timestamp); CREATE INDEX IF NOT EXISTS idx_events_timestamp ON site_events(timestamp);
CREATE INDEX IF NOT EXISTS idx_events_type ON site_events(event_type);
CREATE INDEX IF NOT EXISTS idx_events_instance ON site_events(instance_id); CREATE INDEX IF NOT EXISTS idx_events_instance ON site_events(instance_id);
CREATE INDEX IF NOT EXISTS idx_events_severity ON site_events(severity); DROP INDEX IF EXISTS idx_events_type;
DROP INDEX IF EXISTS idx_events_severity;
CREATE INDEX IF NOT EXISTS idx_events_type_timestamp ON site_events(event_type, timestamp);
CREATE INDEX IF NOT EXISTS idx_events_severity_timestamp ON site_events(severity, timestamp);
"""; """;
// The type and severity indexes are composites with timestamp (#123). The query
// service always orders by (timestamp, id) and returns one page, so a
// single-column index made an equality filter read EVERY matching row and sort
// it before returning 500 (severity=Error over a full 1 GB log: 2.3 s cold,
// 115 ms warm). The composite returns the rows already in timestamp order and
// stops at the page limit (~1 ms). The cost is ~69 bytes per row more index,
// which the storage-cap size model charges (EventLogPurgeService sampling SQL).
// The DROPs migrate a pre-#123 file in place on its first boot; dropping and
// building an index on a 1 GB table takes seconds and runs once. Indexes are not
// part of LocalDb's replicated-table digest, so a pair may upgrade one node at a
// time as far as this change is concerned.
//
// The query service also supports keyword search via leading-wildcard // The query service also supports keyword search via leading-wildcard
// LIKE on message/source. A leading-wildcard LIKE cannot use a B-tree // LIKE on message/source. A leading-wildcard LIKE cannot use a B-tree
// index, so that path intentionally full-scans; severity/event_type/ // index, so that path intentionally full-scans; severity/event_type/
@@ -8,9 +8,11 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests;
/// <summary> /// <summary>
/// Query-shape coverage for the Notification Outbox KPI snapshots (arch-review 04). /// Query-shape coverage for the Notification Outbox KPI snapshots (arch-review 04).
/// Each snapshot must be a single-query conditional aggregation with a /// Each snapshot is at most three status-equality legs (live queue, parked, delivered
/// bounded oldest-pending — the old shape issued 5-7 sequential COUNT round trips /// in the interval) — the shape before arch-review 04 issued 5-7 sequential COUNT round
/// and materialized the whole live queue to find the oldest row. /// 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> /// </summary>
public class NotificationOutboxRepositoryKpiQueryShapeTests public class NotificationOutboxRepositoryKpiQueryShapeTests
{ {
@@ -25,10 +27,10 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
SourceNode = sourceNode, 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> /// <returns>A task that represents the asynchronous operation.</returns>
[Fact] [Fact]
public async Task AllThreeKpiSnapshots_IssueAtMostTwoQueries_AndBoundTheOldestLookup() public async Task AllThreeKpiSnapshots_IssueAtMostThreeStatusLegs_NeverOneOrdScan()
{ {
var counter = new SelectCountingInterceptor(); var counter = new SelectCountingInterceptor();
await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter); await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter);
@@ -45,13 +47,12 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
var stuckCutoff = now.AddMinutes(-5); var stuckCutoff = now.AddMinutes(-5);
var deliveredSince = now.AddMinutes(-60); 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(); counter.Reset();
var global = await repo.ComputeKpisAsync(stuckCutoff, deliveredSince); var global = await repo.ComputeKpisAsync(stuckCutoff, deliveredSince);
Assert.True(counter.Count <= 2, Assert.True(counter.Count <= 3,
$"ComputeKpisAsync issued {counter.Count} queries against Notifications; expected <= 2"); $"ComputeKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
// The oldest lookup must be bounded (LIMIT/TOP), never a full non-terminal SELECT. AssertNoLegMixesParkedWithLive(counter.Commands);
Assert.Contains(counter.Commands, sql => sql.Contains("LIMIT", StringComparison.OrdinalIgnoreCase));
// The aggregation must be PREDICATE-RESTRICTED, not an unrestricted // The aggregation must be PREDICATE-RESTRICTED, not an unrestricted
// scan. Every KPI here concerns the live queue, the parked backlog or the // scan. Every KPI here concerns the live queue, the parked backlog or the
// last delivery interval; historical Delivered/Discarded rows — which are // 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). // Per-site: a single grouped conditional-aggregation query (counts + MIN).
counter.Reset(); counter.Reset();
var perSite = await repo.ComputePerSiteKpisAsync(stuckCutoff, deliveredSince); var perSite = await repo.ComputePerSiteKpisAsync(stuckCutoff, deliveredSince);
Assert.True(counter.Count <= 2, Assert.True(counter.Count <= 3,
$"ComputePerSiteKpisAsync issued {counter.Count} queries against Notifications; expected <= 2"); $"ComputePerSiteKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
AssertNoLegMixesParkedWithLive(counter.Commands);
var siteA = Assert.Single(perSite); var siteA = Assert.Single(perSite);
Assert.Equal(3, siteA.QueueDepth); Assert.Equal(3, siteA.QueueDepth);
Assert.NotNull(siteA.OldestPendingAge); Assert.NotNull(siteA.OldestPendingAge);
@@ -77,13 +79,49 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
// Per-node: same single grouped aggregation, keyed by SourceNode. // Per-node: same single grouped aggregation, keyed by SourceNode.
counter.Reset(); counter.Reset();
var perNode = await repo.ComputePerNodeKpisAsync(stuckCutoff, deliveredSince); var perNode = await repo.ComputePerNodeKpisAsync(stuckCutoff, deliveredSince);
Assert.True(counter.Count <= 2, Assert.True(counter.Count <= 3,
$"ComputePerNodeKpisAsync issued {counter.Count} queries against Notifications; expected <= 2"); $"ComputePerNodeKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
AssertNoLegMixesParkedWithLive(counter.Commands);
var nodeA = Assert.Single(perNode); var nodeA = Assert.Single(perNode);
Assert.Equal(3, nodeA.QueueDepth); Assert.Equal(3, nodeA.QueueDepth);
Assert.NotNull(nodeA.OldestPendingAge); 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> /// <summary>
/// Records the text of every reader command touching the Notifications table so a /// 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. /// 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()); 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> /// <summary>Verifies that duplicate event ids within one batch collapse to a single row.</summary>
/// <returns>A task that represents the asynchronous operation.</returns> /// <returns>A task that represents the asynchronous operation.</returns>
[SkippableFact] [SkippableFact]
@@ -47,7 +47,7 @@ public class SchemaIndexTests : IDisposable
using var reader = cmd.ExecuteReader(); using var reader = cmd.ExecuteReader();
while (reader.Read()) indexes.Add(reader.GetString(0)); while (reader.Read()) indexes.Add(reader.GetString(0));
Assert.Contains("idx_events_severity", indexes); Assert.Contains("idx_events_severity_timestamp", indexes);
} }
/// <summary>Verifies that severity filtered query, when uses index, not full scan.</summary> /// <summary>Verifies that severity filtered query, when uses index, not full scan.</summary>
@@ -70,7 +70,75 @@ public class SchemaIndexTests : IDisposable
} }
var planText = plan.ToString(); var planText = plan.ToString();
Assert.Contains("idx_events_severity", planText); Assert.Contains("idx_events_severity_timestamp", planText);
Assert.DoesNotContain("SCAN site_events", planText); Assert.DoesNotContain("SCAN site_events", planText);
} }
/// <summary>
/// Verifies that the query service's own page shape — an equality filter, ordered by
/// (timestamp, id), limited to a page — is answered in index order rather than by
/// sorting every matching row (#123).
/// </summary>
/// <param name="column">The equality-filtered column.</param>
/// <param name="expectedIndex">The composite index expected to answer it.</param>
[Theory]
[InlineData("severity", "idx_events_severity_timestamp")]
[InlineData("event_type", "idx_events_type_timestamp")]
public void EqualityFilteredPage_ReadsInTimestampOrder_WithoutSortingEveryMatch(string column, string expectedIndex)
{
foreach (var window in new[] { "", "AND timestamp >= '2026-01-01' " })
{
using var cmd = _verifyConnection.CreateCommand();
cmd.CommandText =
$"EXPLAIN QUERY PLAN SELECT id FROM site_events WHERE {column} = 'x' {window}" +
"ORDER BY timestamp ASC, id ASC LIMIT 501";
var plan = new System.Text.StringBuilder();
using var reader = cmd.ExecuteReader();
while (reader.Read())
plan.Append(reader.GetString(reader.GetOrdinal("detail"))).Append('\n');
var planText = plan.ToString();
Assert.Contains(expectedIndex, planText);
// "USE TEMP B-TREE FOR LAST TERM OF ORDER BY" (ties on timestamp, broken by id)
// is fine; a whole-ORDER-BY temp b-tree means every match is read and sorted.
Assert.DoesNotContain("USE TEMP B-TREE FOR ORDER BY", planText);
}
}
/// <summary>Verifies that a pre-#123 file loses its single-column type/severity indexes on the next schema apply.</summary>
[Fact]
public void Apply_OnAPre123File_ReplacesTheSingleColumnIndexes()
{
using var connection = new SqliteConnection("Data Source=:memory:");
connection.Open();
using (var cmd = connection.CreateCommand())
{
cmd.CommandText = """
CREATE TABLE site_events (id TEXT NOT NULL PRIMARY KEY, timestamp TEXT NOT NULL,
event_type TEXT NOT NULL, severity TEXT NOT NULL, instance_id TEXT,
source TEXT NOT NULL, message TEXT NOT NULL, details TEXT);
CREATE INDEX idx_events_timestamp ON site_events(timestamp);
CREATE INDEX idx_events_type ON site_events(event_type);
CREATE INDEX idx_events_instance ON site_events(instance_id);
CREATE INDEX idx_events_severity ON site_events(severity);
INSERT INTO site_events VALUES ('a', '2026-01-01T00:00:00.0000000+00:00', 'alarm', 'Info', NULL, 's', 'm', NULL);
""";
cmd.ExecuteNonQuery();
}
SiteEventLogSchema.Apply(connection);
using var list = connection.CreateCommand();
list.CommandText = "SELECT name FROM pragma_index_list('site_events') WHERE origin = 'c' ORDER BY name";
var indexes = new List<string>();
using (var reader = list.ExecuteReader())
while (reader.Read()) indexes.Add(reader.GetString(0));
Assert.Equal(
["idx_events_instance", "idx_events_severity_timestamp", "idx_events_timestamp", "idx_events_type_timestamp"],
indexes);
using var count = connection.CreateCommand();
count.CommandText = "SELECT COUNT(*) FROM site_events INDEXED BY idx_events_severity_timestamp WHERE severity = 'Info'";
Assert.Equal(1L, (long)count.ExecuteScalar()!);
}
} }
@@ -121,7 +121,7 @@ public class SiteEventLogSchemaTests
while (reader.Read()) indexes.Add(reader.GetString(0)); while (reader.Read()) indexes.Add(reader.GetString(0));
Assert.Equal( Assert.Equal(
["idx_events_instance", "idx_events_severity", "idx_events_timestamp", "idx_events_type"], ["idx_events_instance", "idx_events_severity_timestamp", "idx_events_timestamp", "idx_events_type_timestamp"],
indexes); indexes);
} }
@@ -76,35 +76,36 @@ public class SiteEventsSchemaCalibrationTests
indexes.SequenceEqual( indexes.SequenceEqual(
[ [
"idx_events_instance", "idx_events_instance",
"idx_events_severity", "idx_events_severity_timestamp",
"idx_events_timestamp", "idx_events_timestamp",
"idx_events_type", "idx_events_type_timestamp",
"sqlite_autoindex_site_events_1", "sqlite_autoindex_site_events_1",
]), ]),
$"{RecalibrationNotice} Actual indexes: {string.Join(", ", indexes)}"); $"{RecalibrationNotice} Actual indexes: {string.Join(", ", indexes)}");
} }
/// <summary>Verifies that each index covers exactly the column the sampler sums.</summary> /// <summary>Verifies that each index covers exactly the columns the sampler sums for it.</summary>
/// <param name="indexName">The index name.</param> /// <param name="indexName">The index name.</param>
/// <param name="expectedColumn">The column the index is expected to cover.</param> /// <param name="expectedColumns">The columns the index is expected to cover, comma-separated, in key order.</param>
[Theory] [Theory]
[InlineData("idx_events_timestamp", "timestamp")] [InlineData("idx_events_timestamp", "timestamp")]
[InlineData("idx_events_type", "event_type")]
[InlineData("idx_events_instance", "instance_id")] [InlineData("idx_events_instance", "instance_id")]
[InlineData("idx_events_severity", "severity")] [InlineData("idx_events_type_timestamp", "event_type,timestamp")]
public void EachIndex_CoversExactlyTheColumnTheSamplerSums(string indexName, string expectedColumn) [InlineData("idx_events_severity_timestamp", "severity,timestamp")]
public void EachIndex_CoversExactlyTheColumnsTheSamplerSums(string indexName, string expectedColumns)
{ {
// The sampler's indexed_bytes expression sums octet_length(timestamp) + // The sampler's indexed_bytes expression sums 3 * octet_length(timestamp) +
// octet_length(event_type) + octet_length(severity) + octet_length(instance_id) — // octet_length(event_type) + octet_length(severity) + octet_length(instance_id):
// one column per index. Widening an index to a composite (or re-pointing it) would // timestamp once for its own index and once for each of the two composites (#123).
// change the real entry width without changing what is sampled. // Widening or re-pointing an index changes the real entry width without changing
// what is sampled.
using var connection = OpenWithSchema(); using var connection = OpenWithSchema();
var columns = ReadStrings( var columns = ReadStrings(
connection, $"SELECT name FROM pragma_index_info('{indexName}')"); connection, $"SELECT name FROM pragma_index_info('{indexName}') ORDER BY seqno");
Assert.True( Assert.True(
columns.SequenceEqual([expectedColumn]), columns.SequenceEqual(expectedColumns.Split(',')),
$"{RecalibrationNotice} Index {indexName} covers: {string.Join(", ", columns)}"); $"{RecalibrationNotice} Index {indexName} covers: {string.Join(", ", columns)}");
} }
} }