merge: observability (register rows 33, 54, 56, 63, 66) — LocalDb backlog KPI history + sustained-threshold badge, honest stream-subscriber gauge rename + real data-connection gauges, central metrics port on the rig, ASPNETCORE_ENVIRONMENT on site containers, import-session rejection counter

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.Commons.Tests/Observability/DataConnectionGaugeTests.cs

@@ -0,0 +1,190 @@
+using System.Diagnostics.Metrics;
+using ZB.MOM.WW.ScadaBridge.Commons.Observability;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+
+namespace ZB.MOM.WW.ScadaBridge.Commons.Tests.Observability;
+
+/// <summary>
+/// Register row 54: the real per-site data-connection gauges. Before these existed the only
+/// connection-ish series was <c>scadabridge_site_connection_up</c>, which counted open
+/// central→site gRPC subscriber streams and sat at 0 on a healthy unwatched site — so there
+/// was no metric that answered "are this site's devices actually connected?" at all.
+/// </summary>
+/// <remarks>
+/// Every assertion here is a DELTA against a baseline taken inside the test, and every
+/// connection name is unique to the test. That is deliberate: the registry behind these
+/// gauges is a process-global static, so absolute assertions would inherit the register
+/// row 32 flake that makes <c>QueueDepthGaugeTests</c> pass alone and fail under the full
+/// concurrent suite. Each test also removes its own connections in a <c>finally</c> so it
+/// leaves the shared registry exactly as it found it.
+/// </remarks>
+public sealed class DataConnectionGaugeTests
+{
+    private const string ConnectedGauge = "scadabridge.site.data_connections.connected";
+    private const string TotalGauge = "scadabridge.site.data_connections.total";
+
+    [Fact]
+    public void ConnectedState_RaisesConnectedAndTotal()
+    {
+        var name = UniqueName();
+        var connectedBefore = ReadGauge(ConnectedGauge);
+        var totalBefore = ReadGauge(TotalGauge);
+
+        try
+        {
+            ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected);
+
+            Assert.Equal(connectedBefore + 1, ReadGauge(ConnectedGauge));
+            Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge));
+        }
+        finally
+        {
+            ScadaBridgeTelemetry.RemoveDataConnection(name);
+        }
+    }
+
+    [Theory]
+    [InlineData(ConnectionHealth.Disconnected)]
+    [InlineData(ConnectionHealth.Connecting)]
+    [InlineData(ConnectionHealth.Error)]
+    public void NonConnectedState_RaisesTotalOnly(ConnectionHealth state)
+    {
+        // The load-bearing shape of the pair: a device that is configured but not talking
+        // must show up in the denominator and NOT in the numerator, because connected < total
+        // is the whole alert condition. A gauge that only counted connected rows would read
+        // identically for "one device down" and "device deleted".
+        var name = UniqueName();
+        var connectedBefore = ReadGauge(ConnectedGauge);
+        var totalBefore = ReadGauge(TotalGauge);
+
+        try
+        {
+            ScadaBridgeTelemetry.RecordDataConnectionState(name, state);
+
+            Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge));
+            Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge));
+        }
+        finally
+        {
+            ScadaBridgeTelemetry.RemoveDataConnection(name);
+        }
+    }
+
+    [Fact]
+    public void TransitionFromConnectedToDisconnected_DropsConnectedButKeepsTotal()
+    {
+        // The DCL calls Record* once per state transition on the SAME name, so a transition
+        // must overwrite rather than accumulate — otherwise a flapping connection would
+        // inflate the total without bound.
+        var name = UniqueName();
+        var connectedBefore = ReadGauge(ConnectedGauge);
+        var totalBefore = ReadGauge(TotalGauge);
+
+        try
+        {
+            ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected);
+            ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Disconnected);
+
+            Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge));
+            Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge));
+        }
+        finally
+        {
+            ScadaBridgeTelemetry.RemoveDataConnection(name);
+        }
+    }
+
+    [Fact]
+    public void RemoveDataConnection_DropsItFromBothGauges()
+    {
+        // An undeployed connection must leave the denominator. If it lingered as a
+        // permanently not-connected row, connected < total would latch on forever and the
+        // alert would be useless after the first undeploy.
+        var name = UniqueName();
+        var connectedBefore = ReadGauge(ConnectedGauge);
+        var totalBefore = ReadGauge(TotalGauge);
+
+        ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected);
+        ScadaBridgeTelemetry.RemoveDataConnection(name);
+
+        Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge));
+        Assert.Equal(totalBefore, ReadGauge(TotalGauge));
+    }
+
+    [Fact]
+    public void RemoveDataConnection_IsIdempotent()
+    {
+        var name = UniqueName();
+        ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected);
+        ScadaBridgeTelemetry.RemoveDataConnection(name);
+
+        var ex = Record.Exception(() => ScadaBridgeTelemetry.RemoveDataConnection(name));
+
+        Assert.Null(ex);
+    }
+
+    [Fact]
+    public void NullOrEmptyConnectionName_IsIgnored()
+    {
+        // Guards the gauges against a malformed caller creating a phantom row that could
+        // never be removed by name.
+        var totalBefore = ReadGauge(TotalGauge);
+
+        ScadaBridgeTelemetry.RecordDataConnectionState(string.Empty, ConnectionHealth.Connected);
+        ScadaBridgeTelemetry.RecordDataConnectionState(null!, ConnectionHealth.Connected);
+
+        Assert.Equal(totalBefore, ReadGauge(TotalGauge));
+    }
+
+    [Fact]
+    public void StreamSubscriberGauge_IsPublishedUnderTheRenamedName()
+    {
+        // Register row 54's rename half. Pins the honest name so a future edit cannot
+        // silently restore scadabridge.site.connection.up, whose meaning (open gRPC
+        // subscriber streams) never matched what operators read it as.
+        var names = PublishedInstrumentNames();
+
+        Assert.Contains("scadabridge.site.stream.subscribers.active", names);
+        Assert.DoesNotContain("scadabridge.site.connection.up", names);
+    }
+
+    private static string UniqueName() => $"conn-{Guid.NewGuid():N}";
+
+    /// <summary>Forces one observation of the named gauge and returns its value.</summary>
+    /// <param name="instrumentName">Instrument name to observe.</param>
+    /// <returns>The observed value, or 0 if the instrument never reported.</returns>
+    private static long ReadGauge(string instrumentName)
+    {
+        long observed = 0;
+        using var listener = new MeterListener();
+        listener.InstrumentPublished = (instrument, l) =>
+        {
+            if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName
+                && instrument.Name == instrumentName)
+            {
+                l.EnableMeasurementEvents(instrument);
+            }
+        };
+        listener.SetMeasurementEventCallback<long>((_, measurement, _, _) => observed = measurement);
+        listener.Start();
+        listener.RecordObservableInstruments();
+        return observed;
+    }
+
+    /// <summary>Collects every instrument name published on the ScadaBridge meter.</summary>
+    /// <returns>The published instrument names.</returns>
+    private static IReadOnlyCollection<string> PublishedInstrumentNames()
+    {
+        var names = new List<string>();
+        using var listener = new MeterListener();
+        listener.InstrumentPublished = (instrument, _) =>
+        {
+            if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName)
+            {
+                names.Add(instrument.Name);
+            }
+        };
+        listener.Start();
+        return names;
+    }
+}

tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs

@@ -443,12 +443,12 @@ public class SiteStreamGrpcServerTests : TestKit
     }
 
     [Fact]
-    public async Task SiteConnectionUpGauge_GoesToOneOnConnect_AndBackToZeroOnCancel()
+    public async Task SiteStreamSubscribersActiveGauge_GoesToOneOnConnect_AndBackToZeroOnCancel()
     {
-        // Telemetry follow-on: the scadabridge.site.connection.up gauge must read
+        // Telemetry follow-on: the scadabridge.site.stream.subscribers.active gauge must read
         // exactly 1 while a site stream is established and return to 0 once the
-        // stream terminates on the cancel path — proving SiteConnectionOpened() is
-        // matched by exactly one SiteConnectionClosed() in the handler's finally.
+        // stream terminates on the cancel path — proving SiteStreamSubscriberOpened() is
+        // matched by exactly one SiteStreamSubscriberClosed() in the handler's finally.
         var server = CreateServer();
         server.SetReady(Sys);
 
@@ -459,7 +459,7 @@ public class SiteStreamGrpcServerTests : TestKit
             listener.InstrumentPublished = (instrument, l) =>
             {
                 if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName &&
-                    instrument.Name == "scadabridge.site.connection.up")
+                    instrument.Name == "scadabridge.site.stream.subscribers.active")
                 {
                     l.EnableMeasurementEvents(instrument);
                 }
@@ -485,7 +485,7 @@ public class SiteStreamGrpcServerTests : TestKit
         // (possibly parallel) tests left behind — read relative so the assertion
         // is robust to test interleaving on the process-wide static counter.
         //
-        // SiteConnectionOpened() runs AFTER the _activeStreams insertion that
+        // SiteStreamSubscriberOpened() runs AFTER the _activeStreams insertion that
         // WaitForConditionAsync(ActiveStreamCount == 1) keys off (see
         // SiteStreamGrpcServer.SubscribeInstance), so a one-shot ReadGauge() here
         // can race the increment under full-suite CPU oversubscription. Poll the
@@ -499,7 +499,7 @@ public class SiteStreamGrpcServerTests : TestKit
         // After the cancel path runs the finally, the gauge is balanced back to
         // the baseline — no leaked "up" count. Poll here too: the gauge decrement
         // and the stream-count drop are independent observations, so under load
-        // give the finally's SiteConnectionClosed() time to land.
+        // give the finally's SiteStreamSubscriberClosed() time to land.
         AwaitAssert(() => Assert.Equal(baseline, ReadGauge()), TimeSpan.FromSeconds(5));
     }
 

tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs

@@ -7,7 +7,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
 /// Telemetry guards for the aggregated live alarm cache (plan #10, Task 6): the
 /// active-aggregator observable gauge reflects Started/Stopped, and the reconnect
 /// counter increments on each recorded reconnect. Mirrors the MeterListener read
-/// pattern used by <c>SiteStreamGrpcServerTests.SiteConnectionUpGauge_*</c>; reads are
+/// pattern used by <c>SiteStreamGrpcServerTests.SiteStreamSubscribersActiveGauge_*</c>; reads are
 /// relative to a baseline so the process-wide static instruments are robust to any
 /// parallel test interleaving.
 /// </summary>

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,
+        };
+}

tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs

@@ -22,8 +22,8 @@ public class ScadaBridgeTelemetryTests
         {
             ScadaBridgeTelemetry.RecordDeploymentApplied();
             ScadaBridgeTelemetry.RecordInboundApiRequest("X");
-            ScadaBridgeTelemetry.SiteConnectionOpened();
-            ScadaBridgeTelemetry.SiteConnectionClosed();
+            ScadaBridgeTelemetry.SiteStreamSubscriberOpened();
+            ScadaBridgeTelemetry.SiteStreamSubscriberClosed();
             ScadaBridgeTelemetry.LiveAlarmAggregatorStarted();
             ScadaBridgeTelemetry.LiveAlarmAggregatorStopped();
             ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();

tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs

@@ -19,6 +19,7 @@ namespace ZB.MOM.WW.ScadaBridge.Transport.Tests.Import;
 public sealed class BundleSessionObservabilityTests
 {
     private const string OpenSessionsGauge = "scadabridge.central.transport.import_sessions.open";
+    private const string RejectedSessionsCounter = "scadabridge.central.transport.import_sessions.rejected";
 
     // ------------------------------------------------------------------
     // Warning log on cap rejection
@@ -83,6 +84,108 @@ public sealed class BundleSessionObservabilityTests
         Assert.Empty(logger.Entries);
     }
 
+    // ------------------------------------------------------------------
+    // Rejection counter (register row 66)
+    // ------------------------------------------------------------------
+
+    [Fact]
+    public void Open_AtCap_IncrementsTheRejectionCounter()
+    {
+        // Row 53 shipped the gauge + Warning; a log line is not alertable. This counter is
+        // the series an operator can page on, so it must move on a genuine cap hit.
+        var clock = new TestClock(DateTimeOffset.UtcNow);
+        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
+        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
+        var future = clock.GetUtcNow().AddMinutes(30);
+
+        for (var i = 0; i < 3; i++)
+        {
+            sut.Open(SessionExpiringAt(future));
+        }
+
+        var delta = ReadRejectionCounter(() =>
+            Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future))));
+
+        Assert.Equal(1, delta);
+    }
+
+    [Fact]
+    public void Open_RepeatedCapHits_AccumulateOnTheRejectionCounter()
+    {
+        // A counter that saturated at 1 would hide exactly the case it exists for: a
+        // sustained rejection rate is what distinguishes real contention from a one-off.
+        var clock = new TestClock(DateTimeOffset.UtcNow);
+        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 1 });
+        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
+        var future = clock.GetUtcNow().AddMinutes(30);
+        sut.Open(SessionExpiringAt(future));
+
+        var delta = ReadRejectionCounter(() =>
+        {
+            for (var i = 0; i < 3; i++)
+            {
+                Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future)));
+            }
+        });
+
+        Assert.Equal(3, delta);
+    }
+
+    [Fact]
+    public void Open_UnderCap_DoesNotIncrementTheRejectionCounter()
+    {
+        // Negative control: the counter is a rejection signal, not an open-session signal.
+        var clock = new TestClock(DateTimeOffset.UtcNow);
+        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
+        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
+
+        var delta = ReadRejectionCounter(() =>
+            sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30))));
+
+        Assert.Equal(0, delta);
+    }
+
+    [Fact]
+    public void Open_AtCapOfExpiredSessions_ReclaimsWithoutIncrementingTheRejectionCounter()
+    {
+        // The load-bearing negative control. The cap check runs TWICE — once before
+        // EvictExpired and once after — and only the post-eviction check is a rejection.
+        // Counting the first would make the series fire routinely on any node whose sweep
+        // has not yet run, turning a page-worthy signal into background noise.
+        var clock = new TestClock(DateTimeOffset.UtcNow);
+        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
+        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
+
+        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
+        sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
+        clock.Advance(TimeSpan.FromMinutes(10));
+
+        var delta = ReadRejectionCounter(() =>
+            sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30))));
+
+        Assert.Equal(0, delta);
+    }
+
+    [Fact]
+    public void Open_ReopeningAnExistingSessionIdAtCap_DoesNotIncrementTheRejectionCounter()
+    {
+        // Re-Opening an id already in the store skips the cap block entirely and overwrites
+        // in place. It consumes no new slot and turns nobody away, so it must not count.
+        var clock = new TestClock(DateTimeOffset.UtcNow);
+        var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
+        var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
+        var future = clock.GetUtcNow().AddMinutes(30);
+
+        var first = SessionExpiringAt(future);
+        sut.Open(first);
+        sut.Open(SessionExpiringAt(future));
+
+        // Store is now exactly at the cap of 2.
+        var delta = ReadRejectionCounter(() => sut.Open(first));
+
+        Assert.Equal(0, delta);
+    }
+
     // ------------------------------------------------------------------
     // OpenSessionCount / gauge
     // ------------------------------------------------------------------
@@ -197,6 +300,39 @@ public sealed class BundleSessionObservabilityTests
         return observed;
     }
 
+    /// <summary>
+    /// Runs <paramref name="act"/> with a listener attached to the import-session rejection
+    /// counter and returns the number of rejections recorded DURING that action.
+    /// </summary>
+    /// <remarks>
+    /// Deliberately a delta over a scoped action rather than an absolute read. The counter
+    /// is a process-global static, so an absolute assertion would inherit the register
+    /// row 32 flake that makes <c>QueueDepthGaugeTests</c> fail only under the full
+    /// concurrent suite. A counter measured this way is immune: the listener exists only
+    /// for the duration of <paramref name="act"/>, and every increment is driven
+    /// synchronously by this test's own <c>Open</c> call on its own store instance, so a
+    /// concurrently-running test class cannot contribute a measurement to this window.
+    /// </remarks>
+    /// <param name="act">The action to measure.</param>
+    /// <returns>Rejections counted while <paramref name="act"/> ran.</returns>
+    private static long ReadRejectionCounter(Action act)
+    {
+        long delta = 0;
+        using var listener = new MeterListener();
+        listener.InstrumentPublished = (instrument, l) =>
+        {
+            if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName
+                && instrument.Name == RejectedSessionsCounter)
+            {
+                l.EnableMeasurementEvents(instrument);
+            }
+        };
+        listener.SetMeasurementEventCallback<long>((_, measurement, _, _) => delta += measurement);
+        listener.Start();
+        act();
+        return delta;
+    }
+
     private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new()
     {
         SessionId = Guid.NewGuid(),