fix(site-call-audit): KPI snapshots as single conditional-aggregate queries (review 2026-10-21 06/PERF-3, 08/P3)

This commit is contained in:
Joseph Doherty
2026-10-22 09:23:53 -04:00
parent 811a561885
commit e7fb71c2de
3 changed files with 240 additions and 145 deletions
@@ -209,6 +209,13 @@ mirroring the Notification Outbox KPI shape:
- Stuck count — `Pending`/`Retrying` older than a configurable threshold
(default 10 minutes); display-only, no escalation.
Each snapshot (global, per-site, per-node) is **one conditional-aggregation
query** over the rows that contribute to at least one metric (non-terminal,
Parked, or Delivered/Failed inside the interval), with the oldest-pending age a
`MIN` over a `CASE` in the same scan — the same shape as the Notification
Outbox repository. (It used to cost 7 / 6 / 6 sequential round trips, paid by
the 10 s dashboard poll and the per-minute KPI sampler.)
## Retention
Daily purge of terminal rows (`Delivered`, `Failed`, `Discarded`) after a
@@ -415,47 +415,55 @@ OPTION (RECOMPILE);";
private const string StatusDelivered = "Delivered";
private const string StatusFailed = "Failed";
/// <summary>
/// Restricts a KPI aggregation to the rows that contribute to at least one metric:
/// every non-terminal row (buffered / stuck / oldest-pending), every Parked row, and
/// Delivered/Failed rows that went terminal inside the interval. The historical
/// terminal bulk of the table contributes to none of them, so the aggregation's cost
/// tracks the working set rather than retention, and a site/node with only uncounted
/// rows does not surface as an all-zero per-scope row (preserving the old
/// union-of-metric-keys semantics). The filtered <c>IX_SiteCalls_NonTerminal</c> /
/// <c>IX_SiteCalls_Terminal</c> indexes cover the legs.
/// </summary>
private IQueryable<SiteCall> KpiContributingRows(DateTime intervalSince) =>
_context.SiteCalls.Where(s => s.TerminalAtUtc == null
|| s.Status == StatusParked
|| ((s.Status == StatusDelivered || s.Status == StatusFailed)
&& s.TerminalAtUtc >= intervalSince));
/// <inheritdoc />
public async Task<SiteCallKpiSnapshot> ComputeKpisAsync(
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default)
{
var now = DateTime.UtcNow;
var bufferedCount = await _context.SiteCalls
.CountAsync(s => s.TerminalAtUtc == null, ct);
var parkedCount = await _context.SiteCalls
.CountAsync(s => s.Status == StatusParked, ct);
var failedLastInterval = await _context.SiteCalls
.CountAsync(s => s.Status == StatusFailed
&& s.TerminalAtUtc != null
&& s.TerminalAtUtc >= intervalSince, ct);
var deliveredLastInterval = await _context.SiteCalls
.CountAsync(s => s.Status == StatusDelivered
&& s.TerminalAtUtc != null
&& s.TerminalAtUtc >= intervalSince, ct);
var stuckCount = await _context.SiteCalls
.CountAsync(s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff, ct);
var nonTerminal = _context.SiteCalls.Where(s => s.TerminalAtUtc == null);
TimeSpan? oldestPendingAge = null;
if (await nonTerminal.AnyAsync(ct))
{
var oldestCreatedAt = await nonTerminal.MinAsync(s => s.CreatedAtUtc, ct);
oldestPendingAge = now - oldestCreatedAt;
}
// One conditional-aggregation pass replaces the five sequential COUNTs + Any +
// Min round trips (review 2026-10-21 06/PERF-3, 08/P3), mirroring
// NotificationOutboxRepository.ComputeKpisAsync. GroupBy(_ => 1) yields a single
// group (no contributing rows → no group → all zeros). The oldest non-terminal
// CreatedAtUtc is a MIN over a CASE in the same scan.
var counts = await KpiContributingRows(intervalSince)
.GroupBy(_ => 1)
.Select(g => new
{
Buffered = g.Count(s => s.TerminalAtUtc == null),
Parked = g.Count(s => s.Status == StatusParked),
Failed = g.Count(s => s.Status == StatusFailed
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
Delivered = g.Count(s => s.Status == StatusDelivered
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
Stuck = g.Count(s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff),
OldestCreatedAt = g.Min(s => s.TerminalAtUtc == null ? (DateTime?)s.CreatedAtUtc : null),
})
.FirstOrDefaultAsync(ct);
return new SiteCallKpiSnapshot(
BufferedCount: bufferedCount,
ParkedCount: parkedCount,
FailedLastInterval: failedLastInterval,
DeliveredLastInterval: deliveredLastInterval,
OldestPendingAge: oldestPendingAge,
StuckCount: stuckCount);
BufferedCount: counts?.Buffered ?? 0,
ParkedCount: counts?.Parked ?? 0,
FailedLastInterval: counts?.Failed ?? 0,
DeliveredLastInterval: counts?.Delivered ?? 0,
OldestPendingAge: counts?.OldestCreatedAt is { } oldest ? now - oldest : null,
StuckCount: counts?.Stuck ?? 0);
}
/// <inheritdoc />
@@ -464,45 +472,35 @@ OPTION (RECOMPILE);";
{
var now = DateTime.UtcNow;
var buffered = await CountBySiteAsync(s => s.TerminalAtUtc == null, ct);
// One grouped conditional-aggregation query (counts + MIN) instead of five
// grouped COUNTs + a grouped MIN (review 2026-10-21 06/PERF-3, 08/P3).
var rows = await KpiContributingRows(intervalSince)
.GroupBy(s => s.SourceSite)
.Select(g => new
{
Site = g.Key,
Buffered = g.Count(s => s.TerminalAtUtc == null),
Parked = g.Count(s => s.Status == StatusParked),
Failed = g.Count(s => s.Status == StatusFailed
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
Delivered = g.Count(s => s.Status == StatusDelivered
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
Stuck = g.Count(s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff),
OldestCreatedAt = g.Min(s => s.TerminalAtUtc == null ? (DateTime?)s.CreatedAtUtc : null),
})
.ToListAsync(ct);
var parked = await CountBySiteAsync(s => s.Status == StatusParked, ct);
var failed = await CountBySiteAsync(
s => s.Status == StatusFailed
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince, ct);
var delivered = await CountBySiteAsync(
s => s.Status == StatusDelivered
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince, ct);
var stuck = await CountBySiteAsync(
s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff, ct);
// Oldest non-terminal CreatedAtUtc per site — a server-side GROUP BY MIN.
var oldest = (await _context.SiteCalls
.Where(s => s.TerminalAtUtc == null)
.GroupBy(s => s.SourceSite)
.Select(g => new { Site = g.Key, Oldest = g.Min(s => s.CreatedAtUtc) })
.ToListAsync(ct))
.ToDictionary(x => x.Site, x => x.Oldest);
var siteIds = buffered.Keys
.Concat(parked.Keys).Concat(failed.Keys)
.Concat(delivered.Keys).Concat(stuck.Keys)
.Distinct()
.OrderBy(s => s, StringComparer.Ordinal);
return siteIds.Select(site => new SiteCallSiteKpiSnapshot(
SourceSite: site,
BufferedCount: buffered.GetValueOrDefault(site),
ParkedCount: parked.GetValueOrDefault(site),
FailedLastInterval: failed.GetValueOrDefault(site),
DeliveredLastInterval: delivered.GetValueOrDefault(site),
OldestPendingAge: oldest.TryGetValue(site, out var createdAt)
? now - createdAt
: null,
StuckCount: stuck.GetValueOrDefault(site))).ToList();
return rows
.OrderBy(r => r.Site, StringComparer.Ordinal)
.Select(r => new SiteCallSiteKpiSnapshot(
SourceSite: r.Site,
BufferedCount: r.Buffered,
ParkedCount: r.Parked,
FailedLastInterval: r.Failed,
DeliveredLastInterval: r.Delivered,
OldestPendingAge: r.OldestCreatedAt is { } oldest ? now - oldest : null,
StuckCount: r.Stuck))
.ToList();
}
/// <inheritdoc />
@@ -511,81 +509,37 @@ OPTION (RECOMPILE);";
{
var now = DateTime.UtcNow;
// Exclude rows with NULL SourceNode — per-node KPIs are only meaningful
// when the node identity is known. Each predicate guards n.SourceNode != null
// so the GROUP BY key is always non-null.
var buffered = await CountByNodeAsync(
s => s.TerminalAtUtc == null && s.SourceNode != null, ct);
var parked = await CountByNodeAsync(
s => s.Status == StatusParked && s.SourceNode != null, ct);
var failed = await CountByNodeAsync(
s => s.Status == StatusFailed
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince
&& s.SourceNode != null, ct);
var delivered = await CountByNodeAsync(
s => s.Status == StatusDelivered
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince
&& s.SourceNode != null, ct);
var stuck = await CountByNodeAsync(
s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff
&& s.SourceNode != null, ct);
// Oldest non-terminal CreatedAtUtc per node — server-side GROUP BY MIN.
var oldest = (await _context.SiteCalls
.Where(s => s.TerminalAtUtc == null && s.SourceNode != null)
.GroupBy(s => s.SourceNode!)
.Select(g => new { Node = g.Key, Oldest = g.Min(s => s.CreatedAtUtc) })
.ToListAsync(ct))
.ToDictionary(x => x.Node, x => x.Oldest);
var nodeNames = buffered.Keys
.Concat(parked.Keys).Concat(failed.Keys)
.Concat(delivered.Keys).Concat(stuck.Keys)
.Distinct()
.OrderBy(n => n, StringComparer.Ordinal);
return nodeNames.Select(node => new SiteCallNodeKpiSnapshot(
SourceNode: node,
BufferedCount: buffered.GetValueOrDefault(node),
ParkedCount: parked.GetValueOrDefault(node),
FailedLastInterval: failed.GetValueOrDefault(node),
DeliveredLastInterval: delivered.GetValueOrDefault(node),
OldestPendingAge: oldest.TryGetValue(node, out var createdAt)
? now - createdAt
: null,
StuckCount: stuck.GetValueOrDefault(node))).ToList();
}
/// <summary>Counts <c>SiteCalls</c> rows matching <paramref name="predicate"/>, grouped by source site.</summary>
private async Task<Dictionary<string, int>> CountBySiteAsync(
System.Linq.Expressions.Expression<Func<SiteCall, bool>> predicate,
CancellationToken ct)
{
return await _context.SiteCalls
.Where(predicate)
.GroupBy(s => s.SourceSite)
.Select(g => new { Site = g.Key, Count = g.Count() })
.ToDictionaryAsync(x => x.Site, x => x.Count, ct);
}
/// <summary>
/// Counts <c>SiteCalls</c> rows matching <paramref name="predicate"/>, grouped by source node.
/// Only rows with a non-null <c>SourceNode</c> should be included; the predicate is
/// responsible for enforcing that guard.
/// </summary>
private async Task<Dictionary<string, int>> CountByNodeAsync(
System.Linq.Expressions.Expression<Func<SiteCall, bool>> predicate,
CancellationToken ct)
{
return await _context.SiteCalls
.Where(predicate)
// Same single grouped aggregation as ComputePerSiteKpisAsync, keyed by
// SourceNode. Rows with a NULL SourceNode are excluded — per-node KPIs are only
// meaningful when the node identity is known.
var rows = await KpiContributingRows(intervalSince)
.Where(s => s.SourceNode != null)
.GroupBy(s => s.SourceNode!)
.Select(g => new { Node = g.Key, Count = g.Count() })
.ToDictionaryAsync(x => x.Node, x => x.Count, ct);
.Select(g => new
{
Node = g.Key,
Buffered = g.Count(s => s.TerminalAtUtc == null),
Parked = g.Count(s => s.Status == StatusParked),
Failed = g.Count(s => s.Status == StatusFailed
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
Delivered = g.Count(s => s.Status == StatusDelivered
&& s.TerminalAtUtc != null && s.TerminalAtUtc >= intervalSince),
Stuck = g.Count(s => s.TerminalAtUtc == null && s.CreatedAtUtc < stuckCutoff),
OldestCreatedAt = g.Min(s => s.TerminalAtUtc == null ? (DateTime?)s.CreatedAtUtc : null),
})
.ToListAsync(ct);
return rows
.OrderBy(r => r.Node, StringComparer.Ordinal)
.Select(r => new SiteCallNodeKpiSnapshot(
SourceNode: r.Node,
BufferedCount: r.Buffered,
ParkedCount: r.Parked,
FailedLastInterval: r.Failed,
DeliveredLastInterval: r.Delivered,
OldestPendingAge: r.OldestCreatedAt is { } oldest ? now - oldest : null,
StuckCount: r.Stuck))
.ToList();
}
private static int GetRankOrThrow(string status)
@@ -0,0 +1,134 @@
using System.Data.Common;
using Microsoft.EntityFrameworkCore.Diagnostics;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests;
/// <summary>
/// Query-shape coverage for the Site Call Audit KPI snapshots (review 2026-10-21
/// 06/PERF-3 + 08/P3). Each snapshot must be a single conditional-aggregation query
/// (plus at most one bounded oldest-pending lookup) — the old shape issued 7 (global)
/// and 6 (per-site / per-node) sequential round trips. Correctness against the real
/// schema is covered by the MS SQL <c>SiteCallAuditRepositoryTests</c>.
/// </summary>
public class SiteCallAuditRepositoryKpiQueryShapeTests
{
private static SiteCall NewRow(
string site, string? node, string status, DateTime createdAt, DateTime? terminalAt = null) =>
new()
{
TrackedOperationId = TrackedOperationId.New(),
Channel = "ApiOutbound",
Target = "ERP.GetOrder",
SourceSite = site,
SourceNode = node,
Status = status,
RetryCount = 0,
CreatedAtUtc = createdAt,
UpdatedAtUtc = terminalAt ?? createdAt,
TerminalAtUtc = terminalAt,
IngestedAtUtc = createdAt,
};
/// <summary>Verifies that all three KPI snapshots issue at most two queries and return the right counts.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task AllThreeKpiSnapshots_IssueAtMostTwoQueries()
{
var counter = new SelectCountingInterceptor();
await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter);
var now = DateTime.UtcNow;
ctx.SiteCalls.Add(NewRow("plant-a", "node-a", "Attempted", now.AddMinutes(-30)));
ctx.SiteCalls.Add(NewRow("plant-a", "node-a", "Attempted", now.AddMinutes(-1)));
ctx.SiteCalls.Add(NewRow("plant-a", "node-a", "Parked", now.AddMinutes(-40), now.AddMinutes(-35)));
ctx.SiteCalls.Add(NewRow("plant-a", "node-b", "Delivered", now.AddMinutes(-3), now.AddMinutes(-2)));
ctx.SiteCalls.Add(NewRow("plant-b", null, "Failed", now.AddMinutes(-3), now.AddMinutes(-2)));
// Historical terminal row outside the interval — contributes to no metric.
ctx.SiteCalls.Add(NewRow("plant-c", "node-c", "Delivered", now.AddDays(-3), now.AddDays(-3)));
await ctx.SaveChangesAsync();
var repo = new SiteCallAuditRepository(ctx);
var stuckCutoff = now.AddMinutes(-10);
var intervalSince = now.AddMinutes(-5);
counter.Reset();
var global = await repo.ComputeKpisAsync(stuckCutoff, intervalSince);
Assert.True(counter.Count <= 2,
$"ComputeKpisAsync issued {counter.Count} queries against SiteCalls; expected <= 2");
Assert.Equal(2, global.BufferedCount);
Assert.Equal(1, global.ParkedCount);
Assert.Equal(1, global.FailedLastInterval);
Assert.Equal(1, global.DeliveredLastInterval);
Assert.Equal(1, global.StuckCount);
Assert.NotNull(global.OldestPendingAge);
Assert.True(global.OldestPendingAge >= TimeSpan.FromMinutes(29));
counter.Reset();
var perSite = await repo.ComputePerSiteKpisAsync(stuckCutoff, intervalSince);
Assert.True(counter.Count <= 2,
$"ComputePerSiteKpisAsync issued {counter.Count} queries against SiteCalls; expected <= 2");
// plant-c contributes to no metric, so it does not surface as an all-zero row.
Assert.Equal(new[] { "plant-a", "plant-b" }, perSite.Select(s => s.SourceSite).ToArray());
var a = perSite[0];
Assert.Equal((2, 1, 0, 1, 1), (a.BufferedCount, a.ParkedCount, a.FailedLastInterval, a.DeliveredLastInterval, a.StuckCount));
Assert.NotNull(a.OldestPendingAge);
var b = perSite[1];
Assert.Equal((0, 0, 1, 0, 0), (b.BufferedCount, b.ParkedCount, b.FailedLastInterval, b.DeliveredLastInterval, b.StuckCount));
Assert.Null(b.OldestPendingAge);
counter.Reset();
var perNode = await repo.ComputePerNodeKpisAsync(stuckCutoff, intervalSince);
Assert.True(counter.Count <= 2,
$"ComputePerNodeKpisAsync issued {counter.Count} queries against SiteCalls; expected <= 2");
// The NULL-node row and the metric-less node-c are excluded.
Assert.Equal(new[] { "node-a", "node-b" }, perNode.Select(n => n.SourceNode).ToArray());
var na = perNode[0];
Assert.Equal((2, 1, 0, 0, 1), (na.BufferedCount, na.ParkedCount, na.FailedLastInterval, na.DeliveredLastInterval, na.StuckCount));
Assert.NotNull(na.OldestPendingAge);
var nb = perNode[1];
Assert.Equal((0, 0, 0, 1, 0), (nb.BufferedCount, nb.ParkedCount, nb.FailedLastInterval, nb.DeliveredLastInterval, nb.StuckCount));
Assert.Null(nb.OldestPendingAge);
}
/// <summary>
/// Records the text of every reader command touching the SiteCalls table.
/// </summary>
private sealed class SelectCountingInterceptor : DbCommandInterceptor
{
private readonly List<string> _commands = new();
/// <summary>The number of reader commands recorded so far.</summary>
public int Count => _commands.Count;
/// <summary>Clears all recorded commands.</summary>
public void Reset() => _commands.Clear();
private void Record(DbCommand command)
{
if (command.CommandText.Contains("SiteCalls", StringComparison.OrdinalIgnoreCase))
{
_commands.Add(command.CommandText);
}
}
/// <inheritdoc />
public override InterceptionResult<DbDataReader> ReaderExecuting(
DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result)
{
Record(command);
return base.ReaderExecuting(command, eventData, result);
}
/// <inheritdoc />
public override ValueTask<InterceptionResult<DbDataReader>> ReaderExecutingAsync(
DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result,
CancellationToken cancellationToken = default)
{
Record(command);
return base.ReaderExecutingAsync(command, eventData, result, cancellationToken);
}
}
}