feat(health): KPI history + sustained-backlog badge for LocalDb replication (row 33)

LocalDbOplogBacklog was reported and Prometheus-exported with nothing consuming
it: not sampled into KPI history, so a purge-burst spike could not be reviewed
after the fact, and no threshold treatment anywhere. Both halves shipped.

KPI SAMPLING. SiteHealthKpiSampleSource now emits localDbOplogBacklog and
localDbReplicationConnected (0/1), and the backlog is charted on the Health
dashboard's per-site trend panel. Charting is the point: a point-in-time tile
can never answer 'did that spike drain?', which is the only thing separating an
expected purge burst from a stalled peer.

Both are CONDITIONALLY emitted rather than coerced with ?? 0 like the
auditBacklogPending metric beside them. Both health-report fields are nullable
and null means 'no reading' — replication not configured (rig site-b/site-c) or
a failed poll — never zero. A fabricated 0 would render a pair that cannot read
its own oplog, or one deliberately unreplicated, as perfectly healthy once a
minute forever, burying the real series. So a site emits 12 metrics when
unreplicated and 14 when replicated, and a GAP in these two series truthfully
means 'not replicated / not reporting'. Note false and null differ for the
connected flag: a configured link that is DOWN samples as 0; no link samples as
nothing. Both fold as Gauge (summing an oplog depth, or a 0/1 flag, per hour is
meaningless).

THRESHOLD SEMANTICS — sustained, not instantaneous. This is the design, not a
refinement. The site_events retention purge is CDC-captured like any other
write, so the backlog jumps by the deleted batch and drains as the peer acks
it; a bare 'backlog > threshold' badge would fire on every replicated site
every day and be ignored within a week. Healthy is defined by the backlog
RETURNING to ~0, which only a dwell can express. CentralHealthAggregator holds
LocalDbBacklogAboveSince — the instant the backlog first rose above
LocalDbBacklogThreshold (default 1000) — and the badge appears only after
LocalDbBacklogSustainedFor (default 15 min). A report at or below the threshold
clears the anchor; one above leaves the ORIGINAL instant intact, so the window
is sustained rather than sliding.

The null case is deliberately a third branch: a null backlog leaves the anchor
UNCHANGED rather than clearing it. Null is evidence neither of a drain nor of a
backlog, so a node going blind mid-climb must not silently reset its own dwell,
and an unreplicated site (null forever) never accumulates an anchor and can
never be badged.

Validation rejects a non-positive threshold (0 would badge every replicated site
permanently, since an idle backlog already sits at it) and a dwell shorter than
one ReportInterval (satisfiable by a single sample, which collapses the check
back into the bare threshold it replaces).

DOC CORRECTION worth flagging: this row's premise was 'mirror the SiteAuditBacklog
configurable threshold badge'. That badge does not exist — Component-HealthMonitoring.md
described it in two places but no option, no comparison and no markup were ever
implemented. The two stale claims are corrected here rather than left to send the
next reader looking for it; the LocalDb badge is built fresh on the Metrics-stale
badge's pattern, which is a real configurable-threshold badge.

Tests: HealthMonitoring 115 (9 new aggregator tests incl. purge-burst-drains,
unreplicated-never-anchors, null-mid-climb, exact-threshold, stale-report; 5 new
validator tests; 4 new sampler tests incl. the null-semantics and false-vs-null
cases); CentralUI 1042 (5 new badge/chart tests incl. the purge-burst negative
control); Commons Kpi 93; KpiHistory 54.

tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Pages/HealthPageTests.cs

@@ -346,6 +346,103 @@ public class HealthPageTests : BunitContext
         });
     }
 
+    // ------------------------------------------------------------------
+    // Sustained LocalDb replication backlog badge (register row 33)
+    // ------------------------------------------------------------------
+
+    [Fact]
+    public void Renders_ReplicationBacklogBadge_WhenBacklogSustainedPastTheDwell()
+    {
+        SeedStates(new SiteHealthState
+        {
+            SiteId = "site-a",
+            IsOnline = true,
+            LastHeartbeatAt = DateTimeOffset.UtcNow,
+            // Default dwell is 15 minutes; 30 minutes above threshold is well past it.
+            LocalDbBacklogAboveSince = DateTimeOffset.UtcNow.AddMinutes(-30),
+        });
+
+        var cut = RenderHealthPage();
+
+        cut.WaitForAssertion(() =>
+        {
+            Assert.Contains("data-test=\"localdb-backlog-badge\"", cut.Markup);
+            Assert.Contains("Replication backlog", cut.Markup);
+        });
+    }
+
+    [Fact]
+    public void DoesNotRender_ReplicationBacklogBadge_DuringAPurgeBurst()
+    {
+        // THE false-positive this badge is shaped to avoid. The daily site_events purge is
+        // CDC-captured, so the backlog spikes and then drains. Two minutes above the
+        // threshold is a purge burst, not a fault, and must stay silent — otherwise the
+        // badge fires on every replicated site every day and gets ignored.
+        SeedStates(new SiteHealthState
+        {
+            SiteId = "site-a",
+            IsOnline = true,
+            LastHeartbeatAt = DateTimeOffset.UtcNow,
+            LocalDbBacklogAboveSince = DateTimeOffset.UtcNow.AddMinutes(-2),
+        });
+
+        var cut = RenderHealthPage();
+
+        cut.WaitForAssertion(() => Assert.Contains("site-a", cut.Markup));
+        Assert.DoesNotContain("data-test=\"localdb-backlog-badge\"", cut.Markup);
+    }
+
+    [Fact]
+    public void DoesNotRender_ReplicationBacklogBadge_ForAnUnreplicatedSite()
+    {
+        // Rig site-b/site-c report a null backlog forever, so the aggregator never gives them
+        // an anchor. A deliberately-unreplicated site must never be badged.
+        SeedStates(new SiteHealthState
+        {
+            SiteId = "site-b",
+            IsOnline = true,
+            LastHeartbeatAt = DateTimeOffset.UtcNow,
+            LocalDbBacklogAboveSince = null,
+        });
+
+        var cut = RenderHealthPage();
+
+        cut.WaitForAssertion(() => Assert.Contains("site-b", cut.Markup));
+        Assert.DoesNotContain("data-test=\"localdb-backlog-badge\"", cut.Markup);
+    }
+
+    [Fact]
+    public void DoesNotRender_ReplicationBacklogBadge_ForAnOfflineSite()
+    {
+        // An offline site's last-known anchor is stale by definition; the Offline badge is the
+        // signal that matters, and stacking a backlog warning on top is noise.
+        SeedStates(new SiteHealthState
+        {
+            SiteId = "site-a",
+            IsOnline = false,
+            LastHeartbeatAt = DateTimeOffset.UtcNow.AddMinutes(-30),
+            LocalDbBacklogAboveSince = DateTimeOffset.UtcNow.AddMinutes(-30),
+        });
+
+        var cut = RenderHealthPage();
+
+        cut.WaitForAssertion(() => Assert.Contains("site-a", cut.Markup));
+        Assert.DoesNotContain("data-test=\"localdb-backlog-badge\"", cut.Markup);
+    }
+
+    [Fact]
+    public void Renders_LocalDbBacklogTrendChart()
+    {
+        // The retrospective half of row 33: a point-in-time tile can never answer "did that
+        // spike drain?", which is the only question that distinguishes healthy from faulty.
+        SeedSites("site-a");
+
+        var cut = RenderHealthPage();
+
+        cut.WaitForAssertion(() =>
+            Assert.Contains("LocalDb Replication Backlog", cut.Markup));
+    }
+
     [Fact]
     public void Renders_OfflineSince_ForOfflineSiteWithStatusChange()
     {

tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Kpi/KpiMetricAggregationTests.cs

@@ -47,6 +47,12 @@ public class KpiMetricAggregationTests
         (KpiSources.SiteHealth, "disabledInstances"),
         (KpiSources.SiteHealth, "auditBacklogPending"),
         (KpiSources.SiteHealth, "eventLogWriteFailures"),
+        // Register row 33. Both are instantaneous readings, so both fold as Gauge (the
+        // default) — summing an oplog depth per hour would be meaningless, and summing a
+        // 0/1 link flag doubly so. Both are CONDITIONALLY emitted: a site that reports no
+        // reading writes no sample rather than a fabricated 0.
+        (KpiSources.SiteHealth, "localDbOplogBacklog"),
+        (KpiSources.SiteHealth, "localDbReplicationConnected"),
     };
 
     // The (source, metric) pairs that must fold as Rate (sum-per-hour).
@@ -76,6 +82,8 @@ public class KpiMetricAggregationTests
         (KpiSources.SiteHealth, "connectionsDown"),
         (KpiSources.SiteHealth, "sfBufferDepth"),
         (KpiSources.SiteHealth, "deployedInstances"),
+        (KpiSources.SiteHealth, "localDbOplogBacklog"),
+        (KpiSources.SiteHealth, "localDbReplicationConnected"),
     };
 
     public static IEnumerable<object[]> AllEmittedCases() =>

tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthMonitoringOptionsValidatorTests.cs

@@ -93,4 +93,61 @@ public class HealthMonitoringOptionsValidatorTests
         Assert.True(result.Failed);
         Assert.Contains("MetricsStaleTimeout", result.FailureMessage);
     }
+
+    // ------------------------------------------------------------------
+    // LocalDb sustained-backlog badge (register row 33)
+    // ------------------------------------------------------------------
+
+    [Theory]
+    [InlineData(0)]
+    [InlineData(-1)]
+    public void NonPositiveLocalDbBacklogThreshold_IsRejected(long threshold)
+    {
+        // An idle backlog of 0 already sits at or above a zero/negative threshold, so this
+        // would badge every replicated site permanently.
+        var result = Validate(new HealthMonitoringOptions { LocalDbBacklogThreshold = threshold });
+
+        Assert.True(result.Failed);
+        Assert.Contains("LocalDbBacklogThreshold", result.FailureMessage);
+    }
+
+    [Fact]
+    public void ZeroLocalDbBacklogSustainedFor_IsRejected()
+    {
+        var result = Validate(new HealthMonitoringOptions
+        {
+            LocalDbBacklogSustainedFor = TimeSpan.Zero,
+        });
+
+        Assert.True(result.Failed);
+        Assert.Contains("LocalDbBacklogSustainedFor", result.FailureMessage);
+    }
+
+    [Fact]
+    public void LocalDbBacklogSustainedFor_ShorterThanReportInterval_IsRejected()
+    {
+        // A dwell shorter than one report interval can be satisfied by a single sample, which
+        // collapses the sustained check back into the bare threshold it replaced — and that
+        // fires on every daily purge burst.
+        var result = Validate(new HealthMonitoringOptions
+        {
+            ReportInterval = TimeSpan.FromSeconds(30),
+            LocalDbBacklogSustainedFor = TimeSpan.FromSeconds(10),
+        });
+
+        Assert.True(result.Failed);
+        Assert.Contains("LocalDbBacklogSustainedFor", result.FailureMessage);
+    }
+
+    [Fact]
+    public void DefaultSustainedWindow_ComfortablyOutlastsAPurgeDrain()
+    {
+        // Guards the intent of the default rather than its digits: the dwell must be long
+        // enough that an expected purge spike drains inside it, or the badge becomes daily
+        // noise and gets ignored.
+        var options = new HealthMonitoringOptions();
+
+        Assert.True(options.LocalDbBacklogSustainedFor >= TimeSpan.FromMinutes(5));
+        Assert.True(options.LocalDbBacklogSustainedFor >= options.ReportInterval);
+    }
 }

tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/Kpi/SiteHealthKpiSampleSourceTests.cs

@@ -99,11 +99,95 @@ public class SiteHealthKpiSampleSourceTests
         Assert.Equal(9, byMetric["auditBacklogPending"]);
         Assert.Equal(11, byMetric["eventLogWriteFailures"]);
 
-        // All 12 metrics emitted, exactly once each, for the one populated site.
+        // 12 metrics for an UNREPLICATED site — this fixture leaves both LocalDb replication
+        // fields null, and register row 33 emits those two conditionally rather than coercing
+        // null to 0. A replicated site emits 14; see the LocalDb tests below.
         Assert.Equal(12, byMetric.Count);
         Assert.Equal(12, samples.Count);
     }
 
+    // ------------------------------------------------------------------
+    // LocalDb replication metrics (register row 33)
+    // ------------------------------------------------------------------
+
+    [Fact]
+    public async Task CollectAsync_NullLocalDbFields_EmitsNeitherLocalDbMetric()
+    {
+        // THE null-semantics case, and the reason these two are not `?? 0` like the audit
+        // backlog above. Null means "no reading" — an unreplicated site (rig site-b/site-c) or
+        // a failed poll. A fabricated 0 backlog would render a pair that cannot read its own
+        // oplog, or one deliberately not replicated, as perfectly healthy once a minute forever.
+        var report = MinimalReport("site-a") with
+        {
+            LocalDbOplogBacklog = null,
+            LocalDbReplicationConnected = null,
+        };
+        var aggregator = new StubAggregator
+        {
+            States = { ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = report } },
+        };
+
+        var samples = await new SiteHealthKpiSampleSource(aggregator).CollectAsync(CapturedAt);
+
+        Assert.DoesNotContain(samples, s => s.Metric == "localDbOplogBacklog");
+        Assert.DoesNotContain(samples, s => s.Metric == "localDbReplicationConnected");
+        Assert.Equal(12, samples.Count);
+    }
+
+    [Fact]
+    public async Task CollectAsync_ReplicatedSite_EmitsBothLocalDbMetrics()
+    {
+        var report = MinimalReport("site-a") with
+        {
+            LocalDbOplogBacklog = 4242,
+            LocalDbReplicationConnected = true,
+        };
+        var aggregator = new StubAggregator
+        {
+            States = { ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = report } },
+        };
+
+        var samples = await new SiteHealthKpiSampleSource(aggregator).CollectAsync(CapturedAt);
+
+        Assert.Equal(4242, samples.Single(s => s.Metric == "localDbOplogBacklog").Value);
+        Assert.Equal(1, samples.Single(s => s.Metric == "localDbReplicationConnected").Value);
+        Assert.Equal(14, samples.Count);
+    }
+
+    [Fact]
+    public async Task CollectAsync_ZeroBacklog_IsStillEmitted()
+    {
+        // A real, measured 0 is genuinely different from "no reading" and must be recorded —
+        // otherwise a healthy replicated site would be indistinguishable from an unreplicated
+        // one, and the drain half of a purge burst would never appear in the series.
+        var report = MinimalReport("site-a") with { LocalDbOplogBacklog = 0 };
+        var aggregator = new StubAggregator
+        {
+            States = { ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = report } },
+        };
+
+        var samples = await new SiteHealthKpiSampleSource(aggregator).CollectAsync(CapturedAt);
+
+        Assert.Equal(0, samples.Single(s => s.Metric == "localDbOplogBacklog").Value);
+    }
+
+    [Fact]
+    public async Task CollectAsync_DisconnectedReplication_EmitsZeroNotAbsent()
+    {
+        // false and null are different states: a configured link that is DOWN versus no link
+        // at all. Only the former is a fault, and it must show up as a 0 sample rather than a
+        // gap, or a broken link would look exactly like an unreplicated site.
+        var report = MinimalReport("site-a") with { LocalDbReplicationConnected = false };
+        var aggregator = new StubAggregator
+        {
+            States = { ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = report } },
+        };
+
+        var samples = await new SiteHealthKpiSampleSource(aggregator).CollectAsync(CapturedAt);
+
+        Assert.Equal(0, samples.Single(s => s.Metric == "localDbReplicationConnected").Value);
+    }
+
     [Fact]
     public async Task CollectAsync_NullAuditBacklog_EmitsZeroAuditBacklogPending()
     {

tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/LocalDbBacklogSustainedTests.cs

@@ -0,0 +1,176 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+
+namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests;
+
+/// <summary>
+/// Register row 33: the sustained-backlog anchor behind the Health dashboard's
+/// "Replication backlog" badge.
+/// </summary>
+/// <remarks>
+/// The design constraint these tests exist to pin: a bare "backlog &gt; threshold" badge is
+/// unusable, because the daily `site_events` retention purge is CDC-captured like any other
+/// write and jumps the backlog by the deleted batch size before it drains. That spike is
+/// expected and documented. So the aggregator holds the instant the backlog FIRST rose above
+/// the threshold, and the badge only appears once the configured dwell has elapsed — healthy
+/// means the backlog comes back down, which no point-in-time check can express.
+/// </remarks>
+public sealed class LocalDbBacklogSustainedTests
+{
+    private const long Threshold = 1000;
+
+    [Fact]
+    public void BacklogAboveThreshold_StartsTheClock()
+    {
+        var (aggregator, time) = NewAggregator();
+        var start = time.GetUtcNow();
+
+        aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1));
+
+        Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince);
+    }
+
+    [Fact]
+    public void BacklogStayingAbove_KeepsTheOriginalStartInstant()
+    {
+        // The window must be sustained, not sliding: if each above-threshold report reset the
+        // anchor, the dwell would never elapse and the badge could never appear at all.
+        var (aggregator, time) = NewAggregator();
+        var start = time.GetUtcNow();
+
+        aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1));
+        time.Advance(TimeSpan.FromMinutes(10));
+        aggregator.ProcessReport(Report("site-a", 2, backlog: Threshold + 5000));
+
+        Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince);
+    }
+
+    [Fact]
+    public void BacklogAtOrBelowThreshold_ClearsTheClock()
+    {
+        // The purge-burst exit path. A spike that drains must discard the anchor well before
+        // the dwell elapses, so the expected daily burst never badges the site.
+        var (aggregator, time) = NewAggregator();
+
+        aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 50_000));
+        time.Advance(TimeSpan.FromMinutes(2));
+        aggregator.ProcessReport(Report("site-a", 2, backlog: 0));
+
+        Assert.Null(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince);
+    }
+
+    [Fact]
+    public void BacklogExactlyAtThreshold_DoesNotStartTheClock()
+    {
+        // The comparison is strictly greater-than, so the threshold is the highest ACCEPTABLE
+        // depth rather than the lowest alarming one.
+        var (aggregator, _) = NewAggregator();
+
+        aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold));
+
+        Assert.Null(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince);
+    }
+
+    [Fact]
+    public void UnreplicatedSite_NeverAccumulatesAnAnchor()
+    {
+        // Rig site-b/site-c: a null backlog forever. Null means "no reading", never zero, so
+        // it must never start a clock — otherwise every unreplicated site would eventually be
+        // badged for a backlog it does not have.
+        var (aggregator, time) = NewAggregator();
+
+        aggregator.ProcessReport(Report("site-b", 1, backlog: null));
+        time.Advance(TimeSpan.FromHours(6));
+        aggregator.ProcessReport(Report("site-b", 2, backlog: null));
+
+        Assert.Null(aggregator.GetSiteState("site-b")!.LocalDbBacklogAboveSince);
+    }
+
+    [Fact]
+    public void NullBacklogMidClimb_LeavesTheAnchorUntouched()
+    {
+        // A failed poll on a replicated node reports null. That is not evidence the backlog
+        // drained, so it must not clear the clock — a node that goes blind mid-climb would
+        // otherwise silently reset its own dwell and never be badged.
+        var (aggregator, time) = NewAggregator();
+        var start = time.GetUtcNow();
+
+        aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1));
+        time.Advance(TimeSpan.FromMinutes(5));
+        aggregator.ProcessReport(Report("site-a", 2, backlog: null));
+
+        Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince);
+    }
+
+    [Fact]
+    public void FirstReportForANewSite_CanStartTheClock()
+    {
+        // The registration path is a separate branch from the update path; a site whose very
+        // first report is already backlogged must not be missed.
+        var (aggregator, time) = NewAggregator();
+        var start = time.GetUtcNow();
+
+        aggregator.ProcessReport(Report("brand-new", 1, backlog: Threshold + 1));
+
+        Assert.Equal(start, aggregator.GetSiteState("brand-new")!.LocalDbBacklogAboveSince);
+    }
+
+    [Fact]
+    public void ThresholdIsConfigurable()
+    {
+        var (aggregator, _) = NewAggregator(threshold: 10);
+
+        aggregator.ProcessReport(Report("site-a", 1, backlog: 11));
+
+        Assert.NotNull(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince);
+    }
+
+    [Fact]
+    public void StaleReport_DoesNotDisturbTheAnchor()
+    {
+        // Out-of-order reports are rejected wholesale by the sequence-number guard; the anchor
+        // must ride along with that rejection rather than being updated from stale data.
+        var (aggregator, time) = NewAggregator();
+        var start = time.GetUtcNow();
+
+        aggregator.ProcessReport(Report("site-a", 5, backlog: Threshold + 1));
+        time.Advance(TimeSpan.FromMinutes(5));
+        aggregator.ProcessReport(Report("site-a", 2, backlog: 0));
+
+        Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince);
+    }
+
+    private static (CentralHealthAggregator Aggregator, TestTimeProvider Time) NewAggregator(
+        long threshold = Threshold)
+    {
+        var time = new TestTimeProvider(DateTimeOffset.UtcNow);
+        var opts = new HealthMonitoringOptions
+        {
+            OfflineTimeout = TimeSpan.FromSeconds(60),
+            LocalDbBacklogThreshold = threshold,
+        };
+        var aggregator = new CentralHealthAggregator(
+            Options.Create(opts), NullLogger<CentralHealthAggregator>.Instance, time);
+        return (aggregator, time);
+    }
+
+    private static SiteHealthReport Report(string siteId, long seq, long? backlog) =>
+        new(
+            SiteId: siteId,
+            SequenceNumber: seq,
+            ReportTimestamp: DateTimeOffset.UtcNow,
+            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)
+        {
+            LocalDbOplogBacklog = backlog,
+        };
+}