feat(observability): honest stream-subscriber name + real data-connection gauges (row 54)

scadabridge_site_connection_up never counted data connections. It counts open
central->site gRPC subscriber streams (Debug View + the site-wide alarm feed),
so it correctly reads 0 on a healthy site nobody is watching — and the name
invited reading it as device health. That misreading has already cost a live
debugging session: register row 48's 'connection_up is 0 on all six site
nodes' was healthy steady state, not the fault it was diagnosed as.

Two halves, per the row:

RENAME. scadabridge.site.connection.up -> scadabridge.site.stream.subscribers.active
(helpers SiteConnectionOpened/Closed -> SiteStreamSubscriberOpened/Closed), and
the instrument gains the unit:"1" the other gauges already carried. Every
reference updated: the gRPC server, both test suites, Component-Communication.md,
components/Commons.md, docker/README.md and docker/seed-live-probe.sh. A test
now pins the new name AND asserts the old one is gone, so it cannot creep back.

NEW. scadabridge.site.data_connections.connected / .total — the first exported
data-connection signal (the state existed on the health report and in the CLI,
but never as a metric). connected < total is the alert condition; neither
number alone separates 'a device is down' from 'nothing is configured', which
is why the denominator ships with the numerator.

Sourcing is deliberate on two counts. It is fed by the DCL at its four state
transitions rather than derived from the health report, so the metric survives
a dead report pipeline — the exact failure the 'metrics stale' flag exists to
catch, and a derived gauge would go blind at the same moment. And it uses a
static registry reached through ScadaBridgeTelemetry (the SiteConnectionOpened
precedent) rather than a DI-injected holder: the DCL actors already take
ISiteHealthCollector through their constructors, and threading a second
dependency through both would ripple into every test that builds them. It also
keeps SiteHealthCollector untouched, which matters right now — another agent is
concurrently editing that file for the tag-resolution path.

Removal is wired too: an undeployed connection leaves the registry, so it stops
counting toward the total instead of latching connected < total on forever.

Both gauges read 0 on the standby (connections open on the active node only) —
healthy, documented, and called out in the docs so it does not become the next
site_connection_up.

No ObservedMeters allowlist change: all three instruments ride the existing
ZB.MOM.WW.ScadaBridge meter. (An unlisted meter exports nothing, silently.)

Tests: Commons 711 passed (9 new, delta-based against a baseline with unique
per-test connection names, so they do not inherit the register row 32
QueueDepthGaugeTests static-state flake); Communication 717; DCL 319.

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.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();