From 6f5d60070ce347245c52c570c0646d8694db9769 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 09:12:00 -0400 Subject: [PATCH] perf(notification-outbox): single-query KPI aggregation; oldest-pending no longer materializes the live queue --- .../NotificationOutboxRepository.cs | 261 ++++++++---------- ...ationOutboxRepositoryKpiQueryShapeTests.cs | 112 ++++++++ 2 files changed, 220 insertions(+), 153 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs index 93b54773..bc3690a2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs @@ -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 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); } /// @@ -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(); } /// @@ -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(); - } - - /// Counts notification rows matching , grouped by source site. - private async Task> CountBySiteAsync( - System.Linq.Expressions.Expression> 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); - } - - /// - /// Counts notification rows matching , grouped by source node. - /// Only rows with a non-null SourceNode should be included; the predicate is - /// responsible for enforcing that guard. - /// - private async Task> CountByNodeAsync( - System.Linq.Expressions.Expression> 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(); } /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs new file mode 100644 index 00000000..5ca60fe6 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs @@ -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; + +/// +/// 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. +/// +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); + } + + /// + /// 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. + /// + private sealed class SelectCountingInterceptor : DbCommandInterceptor + { + private readonly List _commands = new(); + + public IReadOnlyList 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 ReaderExecuting( + DbCommand command, CommandEventData eventData, InterceptionResult result) + { + Record(command); + return base.ReaderExecuting(command, eventData, result); + } + + public override ValueTask> ReaderExecutingAsync( + DbCommand command, CommandEventData eventData, InterceptionResult result, + CancellationToken cancellationToken = default) + { + Record(command); + return base.ReaderExecutingAsync(command, eventData, result, cancellationToken); + } + } +}