perf(notification-outbox): single-query KPI aggregation; oldest-pending no longer materializes the live queue
This commit is contained in:
Joseph Doherty
+108
-153
@@ -241,45 +241,40 @@ VALUES
|
||||
{
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
|
||||
var queueDepth = await _context.Notifications
|
||||
.CountAsync(n => n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying, cancellationToken);
|
||||
// One conditional-aggregation pass replaces four sequential COUNT round trips:
|
||||
// each metric is a COUNT(CASE WHEN <predicate> THEN 1 END) over the same scan
|
||||
// (arch-review 04). GroupBy(_ => 1) yields a single group (no rows → no group).
|
||||
var counts = await _context.Notifications
|
||||
.GroupBy(_ => 1)
|
||||
.Select(g => new
|
||||
{
|
||||
QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying),
|
||||
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),
|
||||
})
|
||||
.FirstOrDefaultAsync(cancellationToken);
|
||||
|
||||
var stuckCount = await _context.Notifications
|
||||
.CountAsync(n => (n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying)
|
||||
&& n.CreatedAt < stuckCutoff, cancellationToken);
|
||||
|
||||
var parkedCount = await _context.Notifications
|
||||
.CountAsync(n => n.Status == NotificationStatus.Parked, cancellationToken);
|
||||
|
||||
var deliveredLastInterval = await _context.Notifications
|
||||
.CountAsync(n => n.Status == NotificationStatus.Delivered
|
||||
&& n.DeliveredAt != null
|
||||
&& n.DeliveredAt >= deliveredSince, cancellationToken);
|
||||
|
||||
// Oldest non-terminal CreatedAt. The DateTimeOffset value converter makes a SQL
|
||||
// Min aggregate awkward, so order ascending and take the first instead.
|
||||
var nonTerminal = _context.Notifications
|
||||
// Oldest non-terminal CreatedAt — TOP(1) via ordered FirstOrDefault, so the
|
||||
// live queue is never materialized. The nullable cast lets FirstOrDefault
|
||||
// return null (no non-terminal rows) instead of throwing.
|
||||
var oldestCreatedAt = await _context.Notifications
|
||||
.Where(n => n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying);
|
||||
|
||||
TimeSpan? oldestPendingAge = null;
|
||||
if (await nonTerminal.AnyAsync(cancellationToken))
|
||||
{
|
||||
var oldestCreatedAt = await nonTerminal
|
||||
.OrderBy(n => n.CreatedAt)
|
||||
.Select(n => n.CreatedAt)
|
||||
.FirstAsync(cancellationToken);
|
||||
oldestPendingAge = now - oldestCreatedAt;
|
||||
}
|
||||
|| n.Status == NotificationStatus.Retrying)
|
||||
.OrderBy(n => n.CreatedAt)
|
||||
.Select(n => (DateTimeOffset?)n.CreatedAt)
|
||||
.FirstOrDefaultAsync(cancellationToken);
|
||||
|
||||
return new NotificationKpiSnapshot(
|
||||
QueueDepth: queueDepth,
|
||||
StuckCount: stuckCount,
|
||||
ParkedCount: parkedCount,
|
||||
DeliveredLastInterval: deliveredLastInterval,
|
||||
OldestPendingAge: oldestPendingAge);
|
||||
QueueDepth: counts?.QueueDepth ?? 0,
|
||||
StuckCount: counts?.StuckCount ?? 0,
|
||||
ParkedCount: counts?.ParkedCount ?? 0,
|
||||
DeliveredLastInterval: counts?.DeliveredLastInterval ?? 0,
|
||||
OldestPendingAge: oldestCreatedAt is null ? null : now - oldestCreatedAt.Value);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -288,49 +283,47 @@ VALUES
|
||||
{
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
|
||||
var queueDepth = await CountBySiteAsync(
|
||||
n => n.Status == NotificationStatus.Pending || n.Status == NotificationStatus.Retrying,
|
||||
cancellationToken);
|
||||
// One grouped conditional-aggregation query per scope: counts AND the oldest
|
||||
// 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)
|
||||
.Select(g => new
|
||||
{
|
||||
Site = g.Key,
|
||||
QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying),
|
||||
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);
|
||||
|
||||
var stuck = await CountBySiteAsync(
|
||||
n => (n.Status == NotificationStatus.Pending || n.Status == NotificationStatus.Retrying)
|
||||
&& n.CreatedAt < stuckCutoff,
|
||||
cancellationToken);
|
||||
|
||||
var parked = await CountBySiteAsync(
|
||||
n => n.Status == NotificationStatus.Parked, cancellationToken);
|
||||
|
||||
var delivered = await CountBySiteAsync(
|
||||
n => n.Status == NotificationStatus.Delivered
|
||||
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince,
|
||||
cancellationToken);
|
||||
|
||||
// Oldest non-terminal CreatedAt per site. A SQL Min over the DateTimeOffset
|
||||
// converter is awkward (see ComputeKpisAsync), so project the non-terminal
|
||||
// (site, created) pairs — the live queue, which stays bounded — and reduce
|
||||
// in memory.
|
||||
var oldest = (await _context.Notifications
|
||||
.Where(n => n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying)
|
||||
.Select(n => new { n.SourceSiteId, n.CreatedAt })
|
||||
.ToListAsync(cancellationToken))
|
||||
.GroupBy(x => x.SourceSiteId)
|
||||
.ToDictionary(g => g.Key, g => g.Min(x => x.CreatedAt));
|
||||
|
||||
var siteIds = queueDepth.Keys
|
||||
.Concat(stuck.Keys).Concat(parked.Keys).Concat(delivered.Keys)
|
||||
.Distinct()
|
||||
.OrderBy(s => s, StringComparer.Ordinal);
|
||||
|
||||
return siteIds.Select(site => new SiteNotificationKpiSnapshot(
|
||||
SourceSiteId: site,
|
||||
QueueDepth: queueDepth.GetValueOrDefault(site),
|
||||
StuckCount: stuck.GetValueOrDefault(site),
|
||||
ParkedCount: parked.GetValueOrDefault(site),
|
||||
DeliveredLastInterval: delivered.GetValueOrDefault(site),
|
||||
OldestPendingAge: oldest.TryGetValue(site, out var createdAt)
|
||||
? now - createdAt
|
||||
: null)).ToList();
|
||||
return rows
|
||||
.OrderBy(r => r.Site, StringComparer.Ordinal)
|
||||
.Select(r => new SiteNotificationKpiSnapshot(
|
||||
SourceSiteId: r.Site,
|
||||
QueueDepth: r.QueueDepth,
|
||||
StuckCount: r.StuckCount,
|
||||
ParkedCount: r.ParkedCount,
|
||||
DeliveredLastInterval: r.DeliveredLastInterval,
|
||||
OldestPendingAge: r.OldestCreatedAt is null ? null : now - r.OldestCreatedAt.Value))
|
||||
.ToList();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -339,83 +332,45 @@ VALUES
|
||||
{
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
|
||||
// Exclude rows with NULL SourceNode (legacy / unstamped) — per-node KPIs
|
||||
// are only meaningful when the node identity is known.
|
||||
var queueDepth = await CountByNodeAsync(
|
||||
n => (n.Status == NotificationStatus.Pending || n.Status == NotificationStatus.Retrying)
|
||||
&& n.SourceNode != null,
|
||||
cancellationToken);
|
||||
|
||||
var stuck = await CountByNodeAsync(
|
||||
n => (n.Status == NotificationStatus.Pending || n.Status == NotificationStatus.Retrying)
|
||||
&& n.CreatedAt < stuckCutoff
|
||||
&& n.SourceNode != null,
|
||||
cancellationToken);
|
||||
|
||||
var parked = await CountByNodeAsync(
|
||||
n => n.Status == NotificationStatus.Parked && n.SourceNode != null,
|
||||
cancellationToken);
|
||||
|
||||
var delivered = await CountByNodeAsync(
|
||||
n => n.Status == NotificationStatus.Delivered
|
||||
&& n.DeliveredAt != null && n.DeliveredAt >= deliveredSince
|
||||
&& n.SourceNode != null,
|
||||
cancellationToken);
|
||||
|
||||
// Oldest non-terminal CreatedAt per node — same in-memory reduction
|
||||
// pattern as ComputePerSiteKpisAsync (DateTimeOffset converter makes
|
||||
// a SQL Min awkward).
|
||||
var oldest = (await _context.Notifications
|
||||
.Where(n => (n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying)
|
||||
&& n.SourceNode != null)
|
||||
.Select(n => new { n.SourceNode, n.CreatedAt })
|
||||
.ToListAsync(cancellationToken))
|
||||
.GroupBy(x => x.SourceNode!)
|
||||
.ToDictionary(g => g.Key, g => g.Min(x => x.CreatedAt));
|
||||
|
||||
var nodeNames = queueDepth.Keys
|
||||
.Concat(stuck.Keys).Concat(parked.Keys).Concat(delivered.Keys)
|
||||
.Distinct()
|
||||
.OrderBy(n => n, StringComparer.Ordinal);
|
||||
|
||||
return nodeNames.Select(node => new NodeNotificationKpiSnapshot(
|
||||
SourceNode: node,
|
||||
QueueDepth: queueDepth.GetValueOrDefault(node),
|
||||
StuckCount: stuck.GetValueOrDefault(node),
|
||||
ParkedCount: parked.GetValueOrDefault(node),
|
||||
DeliveredLastInterval: delivered.GetValueOrDefault(node),
|
||||
OldestPendingAge: oldest.TryGetValue(node, out var createdAt)
|
||||
? now - createdAt
|
||||
: null)).ToList();
|
||||
}
|
||||
|
||||
/// <summary>Counts notification rows matching <paramref name="predicate"/>, grouped by source site.</summary>
|
||||
private async Task<Dictionary<string, int>> CountBySiteAsync(
|
||||
System.Linq.Expressions.Expression<Func<Notification, bool>> predicate,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await _context.Notifications
|
||||
.Where(predicate)
|
||||
.GroupBy(n => n.SourceSiteId)
|
||||
.Select(g => new { Site = g.Key, Count = g.Count() })
|
||||
.ToDictionaryAsync(x => x.Site, x => x.Count, cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Counts notification 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<Notification, bool>> predicate,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await _context.Notifications
|
||||
.Where(predicate)
|
||||
// Same single-scan grouped aggregation as ComputePerSiteKpisAsync, keyed by
|
||||
// SourceNode. Rows with a NULL SourceNode (legacy / unstamped) are excluded —
|
||||
// per-node KPIs are only meaningful when node identity is known.
|
||||
var rows = await _context.Notifications
|
||||
.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!)
|
||||
.Select(g => new { Node = g.Key, Count = g.Count() })
|
||||
.ToDictionaryAsync(x => x.Node, x => x.Count, cancellationToken);
|
||||
.Select(g => new
|
||||
{
|
||||
Node = g.Key,
|
||||
QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending
|
||||
|| n.Status == NotificationStatus.Retrying),
|
||||
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);
|
||||
|
||||
return rows
|
||||
.OrderBy(r => r.Node, StringComparer.Ordinal)
|
||||
.Select(r => new NodeNotificationKpiSnapshot(
|
||||
SourceNode: r.Node,
|
||||
QueueDepth: r.QueueDepth,
|
||||
StuckCount: r.StuckCount,
|
||||
ParkedCount: r.ParkedCount,
|
||||
DeliveredLastInterval: r.DeliveredLastInterval,
|
||||
OldestPendingAge: r.OldestCreatedAt is null ? null : now - r.OldestCreatedAt.Value))
|
||||
.ToList();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
+112
@@ -0,0 +1,112 @@
|
||||
using System.Data.Common;
|
||||
using Microsoft.EntityFrameworkCore.Diagnostics;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Notifications;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Query-shape coverage for the Notification Outbox KPI snapshots (arch-review 04,
|
||||
/// Task 20). 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.
|
||||
/// </summary>
|
||||
public class NotificationOutboxRepositoryKpiQueryShapeTests
|
||||
{
|
||||
private static Notification NewNotification(
|
||||
string sourceSiteId, string? sourceNode, NotificationStatus status,
|
||||
DateTimeOffset createdAt, DateTimeOffset? deliveredAt = null) =>
|
||||
new(Guid.NewGuid().ToString(), NotificationType.Email, "Ops List", "Subject", "Body", sourceSiteId)
|
||||
{
|
||||
Status = status,
|
||||
CreatedAt = createdAt,
|
||||
DeliveredAt = deliveredAt,
|
||||
SourceNode = sourceNode,
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public async Task AllThreeKpiSnapshots_IssueAtMostTwoQueries_AndBoundTheOldestLookup()
|
||||
{
|
||||
var counter = new SelectCountingInterceptor();
|
||||
await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter);
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
|
||||
// Three non-terminal rows for one site/node — the "live queue" the old code
|
||||
// pulled into memory to find the oldest. All older than the stuck cutoff.
|
||||
ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Pending, now.AddMinutes(-30)));
|
||||
ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Retrying, now.AddMinutes(-20)));
|
||||
ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Pending, now.AddMinutes(-10)));
|
||||
await ctx.SaveChangesAsync();
|
||||
|
||||
var repo = new NotificationOutboxRepository(ctx);
|
||||
var stuckCutoff = now.AddMinutes(-5);
|
||||
var deliveredSince = now.AddMinutes(-60);
|
||||
|
||||
// Global: one aggregation query + one bounded oldest query.
|
||||
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.Equal(3, global.QueueDepth);
|
||||
Assert.NotNull(global.OldestPendingAge);
|
||||
|
||||
// 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");
|
||||
var siteA = Assert.Single(perSite);
|
||||
Assert.Equal(3, siteA.QueueDepth);
|
||||
Assert.NotNull(siteA.OldestPendingAge);
|
||||
|
||||
// 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");
|
||||
var nodeA = Assert.Single(perNode);
|
||||
Assert.Equal(3, nodeA.QueueDepth);
|
||||
Assert.NotNull(nodeA.OldestPendingAge);
|
||||
}
|
||||
|
||||
/// <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.
|
||||
/// </summary>
|
||||
private sealed class SelectCountingInterceptor : DbCommandInterceptor
|
||||
{
|
||||
private readonly List<string> _commands = new();
|
||||
|
||||
public IReadOnlyList<string> Commands => _commands;
|
||||
|
||||
public int Count => _commands.Count;
|
||||
|
||||
public void Reset() => _commands.Clear();
|
||||
|
||||
private void Record(DbCommand command)
|
||||
{
|
||||
if (command.CommandText.Contains("Notifications", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_commands.Add(command.CommandText);
|
||||
}
|
||||
}
|
||||
|
||||
public override InterceptionResult<DbDataReader> ReaderExecuting(
|
||||
DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result)
|
||||
{
|
||||
Record(command);
|
||||
return base.ReaderExecuting(command, eventData, result);
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user