fix(kpi-history): backlogTotal trend records the real site-backlog aggregate instead of a hardwired zero

2026-07-09 07:51:55 -04:00
parent 1dd993ec2f
commit 1b53de1933
6 changed files with 271 additions and 3 deletions
@@ -3,6 +3,12 @@ using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi;
 using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;

+// Task 10 (arch-review 04): the append-only repository leaves the snapshot's
+// BacklogTotal at zero; the real backlog is the cross-site health aggregate that
+// only the central node knows. An optional IAuditBacklogProvider supplies it so
+// the *history* sample stops recording a hardwired zero (the live tile already
+// filled this in). Sites/tests leave it unregistered → null → snapshot fallback.
+
 namespace ZB.MOM.WW.ScadaBridge.AuditLog.Kpi;

 /// <summary>
@@ -44,18 +50,30 @@ public sealed class AuditLogKpiSampleSource : IKpiSampleSource
    private const string BacklogTotalMetric = KpiMetrics.AuditLog.BacklogTotal;

    private readonly IAuditLogRepository _repository;
+    private readonly IAuditBacklogProvider? _backlogProvider;

    /// <summary>
    /// Initializes a new instance of the <see cref="AuditLogKpiSampleSource"/> class.
    /// </summary>
    /// <param name="repository">
    /// Append-only Audit Log repository — its <see cref="IAuditLogRepository.GetKpiSnapshotAsync"/>
-    /// supplies the volume, error, and backlog counts snapshotted here.
+    /// supplies the volume and error counts snapshotted here (its <c>BacklogTotal</c>
+    /// is always zero — see <paramref name="backlogProvider"/>).
    /// </param>
-    public AuditLogKpiSampleSource(IAuditLogRepository repository)
+    /// <param name="backlogProvider">
+    /// Optional cross-site backlog aggregate. When registered (central node), the
+    /// <c>backlogTotal</c> sample records its value; when absent (sites / unit tests,
+    /// where MS.DI resolves the default) the source falls back to the snapshot's own
+    /// <c>BacklogTotal</c> — i.e. zero, the pre-Task-10 behavior. Injecting the seam
+    /// keeps the append-only repo out of the in-memory health aggregator.
+    /// </param>
+    public AuditLogKpiSampleSource(
+        IAuditLogRepository repository,
+        IAuditBacklogProvider? backlogProvider = null)
    {
        ArgumentNullException.ThrowIfNull(repository);
        _repository = repository;
+        _backlogProvider = backlogProvider;
    }

    /// <inheritdoc/>
@@ -78,11 +96,15 @@ public sealed class AuditLogKpiSampleSource : IKpiSampleSource
            return [];
        }

+        // Backlog: the provider's cross-site aggregate when registered (central),
+        // otherwise the snapshot's own value (zero) as the site/test fallback.
+        var backlogTotal = _backlogProvider?.GetPendingBacklogTotal() ?? snapshot.BacklogTotal;
+
        return
        [
            Sample(TotalEventsLastHourMetric, snapshot.TotalEventsLastHour, capturedAtUtc),
            Sample(ErrorEventsLastHourMetric, snapshot.ErrorEventsLastHour, capturedAtUtc),
-            Sample(BacklogTotalMetric, snapshot.BacklogTotal, capturedAtUtc),
+            Sample(BacklogTotalMetric, backlogTotal, capturedAtUtc),
        ];
    }

@@ -0,0 +1,33 @@
+namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi;
+
+/// <summary>
+/// Supplies the system-wide pending Audit Log backlog for the
+/// <c>backlogTotal</c> KPI history sample ("KPI History &amp; Trends").
+/// </summary>
+/// <remarks>
+/// <para>
+/// The append-only Audit Log repository leaves <c>AuditLogKpiSnapshot.BacklogTotal</c>
+/// at zero — the backlog is not a central-table count but the sum of the latest
+/// per-site <c>SiteAuditBacklog.PendingCount</c> health reports, which live only in
+/// the in-memory central health aggregator. The live Health-dashboard "Audit" tile
+/// already fills this in (<c>AuditLogQueryService.GetKpiSnapshotAsync</c>); this seam
+/// lets the KPI <em>history</em> sample source read the same aggregate instead of
+/// persisting a hardwired zero.
+/// </para>
+/// <para>
+/// Best-effort: sites that have not yet reported (or whose reporter is disabled)
+/// contribute zero — a missing snapshot is treated as "unknown, count as nothing",
+/// mirroring the live tile. Central-only: registered on the central node where the
+/// health aggregator lives; sites and unit tests leave it unregistered so the
+/// sample source falls back to the snapshot's own <c>BacklogTotal</c>.
+/// </para>
+/// </remarks>
+public interface IAuditBacklogProvider
+{
+    /// <summary>
+    /// Returns the current total pending (not-yet-forwarded) Audit Log backlog,
+    /// summed across every site's latest health report. Never negative; zero when
+    /// no site is reporting a backlog.
+    /// </summary>
+    long GetPendingBacklogTotal();
+}
@@ -0,0 +1,52 @@
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi;
+
+namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring.Kpi;
+
+/// <summary>
+/// Central <see cref="IAuditBacklogProvider"/> — sums the latest per-site
+/// <c>SiteAuditBacklog.PendingCount</c> across the in-memory
+/// <see cref="ICentralHealthAggregator"/>, the same aggregate the live
+/// Health-dashboard "Audit" backlog tile shows
+/// (<c>AuditLogQueryService.GetKpiSnapshotAsync</c>).
+/// </summary>
+/// <remarks>
+/// Registered only on the central node (where the aggregator lives), so the
+/// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi.IKpiSampleSource"/>
+/// backing the <c>backlogTotal</c> trend records the real backlog instead of the
+/// append-only repository's hardwired zero. Best-effort: sites with no report or
+/// a null <c>SiteAuditBacklog</c> contribute nothing.
+/// </remarks>
+public sealed class CentralHealthAuditBacklogProvider : IAuditBacklogProvider
+{
+    private readonly ICentralHealthAggregator _aggregator;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="CentralHealthAuditBacklogProvider"/> class.
+    /// </summary>
+    /// <param name="aggregator">The central health aggregator whose per-site latest reports carry the backlog.</param>
+    public CentralHealthAuditBacklogProvider(ICentralHealthAggregator aggregator)
+    {
+        ArgumentNullException.ThrowIfNull(aggregator);
+        _aggregator = aggregator;
+    }
+
+    /// <inheritdoc />
+    public long GetPendingBacklogTotal()
+    {
+        // Ported verbatim from AuditLogQueryService.GetKpiSnapshotAsync: a null
+        // SiteAuditBacklog is "unknown" (contributes zero), and the `> 0` guard
+        // ignores any defensive negative/zero value so the tile and the history
+        // agree exactly.
+        long backlog = 0;
+        foreach (var state in _aggregator.GetAllSiteStates().Values)
+        {
+            var pending = state.LatestReport?.SiteAuditBacklog?.PendingCount;
+            if (pending is > 0)
+            {
+                backlog += pending.Value;
+            }
+        }
+
+        return backlog;
+    }
+}
@@ -63,6 +63,15 @@ public static class ServiceCollectionExtensions
        services.AddHostedService(sp => sp.GetRequiredService<CentralHealthAggregator>());
        services.AddHostedService<CentralHealthReportLoop>();

+        // Task 10 (arch-review 04): the central backlog aggregate for the Audit Log
+        // KPI *history*. Only the central node runs the aggregator, so this seam is
+        // registered here; the AuditLogKpiSampleSource resolves it (optional ctor
+        // param) and records the real cross-site backlog instead of the append-only
+        // repository's hardwired zero. Sites leave it unregistered → snapshot fallback.
+        services.AddSingleton<
+            Commons.Interfaces.Kpi.IAuditBacklogProvider,
+            Kpi.CentralHealthAuditBacklogProvider>();
+
        // "KPI History & Trends": per-site Site Health KPI sample source.
        // Reads the in-memory central aggregator (a singleton) rather than a
        // repository; registered Scoped to match the recorder's per-tick scope
@@ -1,5 +1,6 @@
 using NSubstitute;
 using ZB.MOM.WW.ScadaBridge.AuditLog.Kpi;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi;
 using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
 using ZB.MOM.WW.ScadaBridge.Commons.Types;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;
@@ -97,6 +98,57 @@ public class AuditLogKpiSampleSourceTests
        Assert.Empty(samples);
    }

+    [Fact]
+    public async Task CollectAsync_uses_backlog_provider_when_present()
+    {
+        // The repository's snapshot carries the hardwired zero the append-only
+        // repo always emits; the provider carries the real cross-site aggregate.
+        var snapshot = new AuditLogKpiSnapshot(
+            TotalEventsLastHour: 10,
+            ErrorEventsLastHour: 1,
+            BacklogTotal: 0,
+            AsOfUtc: CapturedAt);
+
+        var repo = Substitute.For<IAuditLogRepository>();
+        repo.GetKpiSnapshotAsync(Arg.Any<TimeSpan>(), Arg.Any<DateTime?>(), Arg.Any<CancellationToken>())
+            .Returns(snapshot);
+
+        var backlog = Substitute.For<IAuditBacklogProvider>();
+        backlog.GetPendingBacklogTotal().Returns(42L);
+
+        var sut = new AuditLogKpiSampleSource(repo, backlog);
+
+        var samples = await sut.CollectAsync(CapturedAt);
+
+        // Volume + error metrics still come from the snapshot; backlog is the
+        // provider's real aggregate, not the snapshot's zero.
+        AssertMetric(samples, "totalEventsLastHour", 10);
+        AssertMetric(samples, "errorEventsLastHour", 1);
+        AssertMetric(samples, "backlogTotal", 42);
+    }
+
+    [Fact]
+    public async Task CollectAsync_falls_back_to_snapshot_backlog_when_provider_absent()
+    {
+        // No provider (sites / unit tests): the sample source degrades to the
+        // snapshot's own BacklogTotal — unchanged pre-Task-10 behavior.
+        var snapshot = new AuditLogKpiSnapshot(
+            TotalEventsLastHour: 10,
+            ErrorEventsLastHour: 1,
+            BacklogTotal: 7,
+            AsOfUtc: CapturedAt);
+
+        var repo = Substitute.For<IAuditLogRepository>();
+        repo.GetKpiSnapshotAsync(Arg.Any<TimeSpan>(), Arg.Any<DateTime?>(), Arg.Any<CancellationToken>())
+            .Returns(snapshot);
+
+        var sut = new AuditLogKpiSampleSource(repo, backlogProvider: null);
+
+        var samples = await sut.CollectAsync(CapturedAt);
+
+        AssertMetric(samples, "backlogTotal", 7);
+    }
+
    private static void AssertMetric(
        IReadOnlyList<Commons.Entities.Kpi.KpiSample> samples,
        string metric,
@@ -0,0 +1,100 @@
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
+using ZB.MOM.WW.ScadaBridge.Commons.Types;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+using ZB.MOM.WW.ScadaBridge.HealthMonitoring.Kpi;
+
+namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests.Kpi;
+
+/// <summary>
+/// Task 10 (arch-review 04) coverage for <see cref="CentralHealthAuditBacklogProvider"/>.
+/// The provider sums the latest per-site <c>SiteAuditBacklog.PendingCount</c> across the
+/// in-memory <see cref="ICentralHealthAggregator"/> — the same aggregate the live Audit
+/// backlog tile shows — so the <c>backlogTotal</c> KPI history stops recording a hardwired
+/// zero. Sites with no report or a null backlog snapshot contribute nothing.
+/// </summary>
+public class CentralHealthAuditBacklogProviderTests
+{
+    private static readonly DateTime At = new(2026, 6, 17, 12, 0, 0, DateTimeKind.Utc);
+
+    [Fact]
+    public void GetPendingBacklogTotal_Sums_PendingCount_Across_Sites()
+    {
+        var aggregator = new StubAggregator
+        {
+            States =
+            {
+                ["site-a"] = StateWithBacklog("site-a", 9),
+                ["site-b"] = StateWithBacklog("site-b", 4),
+                ["site-c"] = new SiteHealthState { SiteId = "site-c", LatestReport = null }, // heartbeat-only
+            },
+        };
+
+        var provider = new CentralHealthAuditBacklogProvider(aggregator);
+
+        // 9 + 4; the null-report site contributes nothing.
+        Assert.Equal(13, provider.GetPendingBacklogTotal());
+    }
+
+    [Fact]
+    public void GetPendingBacklogTotal_Returns_Zero_When_No_Backlog_Reported()
+    {
+        var aggregator = new StubAggregator
+        {
+            States =
+            {
+                ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = MinimalReport("site-a") with { SiteAuditBacklog = null } },
+                ["site-b"] = new SiteHealthState { SiteId = "site-b", LatestReport = null },
+            },
+        };
+
+        var provider = new CentralHealthAuditBacklogProvider(aggregator);
+
+        Assert.Equal(0, provider.GetPendingBacklogTotal());
+    }
+
+    private static SiteHealthState StateWithBacklog(string siteId, int pending) =>
+        new()
+        {
+            SiteId = siteId,
+            LatestReport = MinimalReport(siteId) with
+            {
+                SiteAuditBacklog = new SiteAuditBacklogSnapshot(
+                    PendingCount: pending, OldestPendingUtc: null, OnDiskBytes: 0),
+            },
+        };
+
+    private static SiteHealthReport MinimalReport(string siteId) =>
+        new(
+            SiteId: siteId,
+            SequenceNumber: 1,
+            ReportTimestamp: At,
+            DataConnectionStatuses: new Dictionary<string, ConnectionHealth>(),
+            TagResolutionCounts: new Dictionary<string, TagResolutionStatus>(),
+            ScriptErrorCount: 0,
+            AlarmEvaluationErrorCount: 0,
+            StoreAndForwardBufferDepths: new Dictionary<string, int>(),
+            DeadLetterCount: 0,
+            DeployedInstanceCount: 0,
+            EnabledInstanceCount: 0,
+            DisabledInstanceCount: 0);
+
+    /// <summary>
+    /// Hand-rolled <see cref="ICentralHealthAggregator"/> stub — the HealthMonitoring.Tests
+    /// project has no mocking library. Only <see cref="GetAllSiteStates"/> is exercised.
+    /// </summary>
+    private sealed class StubAggregator : ICentralHealthAggregator
+    {
+        public Dictionary<string, SiteHealthState> States { get; } = new();
+
+        public IReadOnlyDictionary<string, SiteHealthState> GetAllSiteStates() => States;
+
+        public SiteHealthState? GetSiteState(string siteId) =>
+            States.TryGetValue(siteId, out var state) ? state : null;
+
+        public void ProcessReport(SiteHealthReport report) => throw new NotSupportedException();
+
+        public void MarkHeartbeat(string siteId, DateTimeOffset receivedAt) => throw new NotSupportedException();
+
+        public void PruneUnknownSites(IReadOnlyCollection<string> knownSiteIds) => throw new NotSupportedException();
+    }
+}