feat(sitestream): validate live-alarm-cache options + active-aggregator/reconnect telemetry (plan #10 T6)

Extend CommunicationOptionsValidator with eager bounds for the four T4
live-alarm-cache options (linger >= 0, reconcile > 0, seed concurrency 1..64,
subscribers-per-site >= 1). Enforce the per-site viewer cap fail-safe in
SiteAlarmLiveCacheService.Subscribe (reject excess viewers with a no-op
disposable rather than growing the list or throwing into the Blazor render
path). Surface two telemetry instruments on the existing ScadaBridgeTelemetry
meter: an active-aggregator observable gauge and a reconnect counter, wired from
the aggregator actor's PreStart/PostStop and its NodeA<->NodeB flip /
reconcile-driven reopen.

Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj

src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs

@@ -47,11 +47,23 @@ public static class ScadaBridgeTelemetry
         Meter.CreateCounter<long>("scadabridge.store_and_forward.replication.failures", unit: "1",
             description: "S&F buffer replication operations that failed to dispatch/deliver to the peer node");
 
+    /// <summary>
+    /// Incremented each time a per-site live-alarm aggregator re-establishes its site-wide
+    /// gRPC stream — a NodeA↔NodeB failover flip or a reconcile-driven reopen after the
+    /// stream was given up (plan #10, Task 6). A sustained climb signals a flapping site link.
+    /// </summary>
+    private static readonly Counter<long> _liveAlarmStreamReconnects =
+        Meter.CreateCounter<long>("scadabridge.site.alarm_cache.reconnects", unit: "1",
+            description: "Live-alarm aggregator site-wide gRPC stream reconnects (NodeA↔NodeB flip or reconcile-driven reopen).");
+
     // ---------------- Observable gauges ----------------
 
     /// <summary>Current count of open site connections, mutated via <see cref="Interlocked"/>.</summary>
     private static long _siteConnectionsUp;
 
+    /// <summary>Current count of running per-site live-alarm aggregators, mutated via <see cref="Interlocked"/>.</summary>
+    private static long _liveAlarmAggregatorsActive;
+
     /// <summary>Provider that yields the live StoreAndForward queue depth; set by a later task.</summary>
     private static Func<long>? _queueDepthProvider;
 
@@ -68,6 +80,13 @@ public static class ScadaBridgeTelemetry
             () => Volatile.Read(ref _queueDepthProvider) is { } p ? p() : 0L,
             unit: "items",
             description: "Current StoreAndForward queue depth.");
+
+    /// <summary>Gauge reporting the number of currently running per-site live-alarm aggregators.</summary>
+    private static readonly ObservableGauge<long> _liveAlarmAggregatorsActiveGauge =
+        Meter.CreateObservableGauge<long>("scadabridge.site.alarm_cache.aggregators.active",
+            () => Interlocked.Read(ref _liveAlarmAggregatorsActive),
+            unit: "1",
+            description: "Number of per-site live-alarm aggregators currently running on the active central node.");
 #pragma warning restore IDE0052
 
     // ---------------- Emit helpers ----------------
@@ -94,6 +113,15 @@ public static class ScadaBridgeTelemetry
     /// <summary>Records that a site connection closed (decrements the up-count gauge).</summary>
     public static void SiteConnectionClosed() => Interlocked.Decrement(ref _siteConnectionsUp);
 
+    /// <summary>Records that a per-site live-alarm aggregator started (increments the active-aggregator gauge).</summary>
+    public static void LiveAlarmAggregatorStarted() => Interlocked.Increment(ref _liveAlarmAggregatorsActive);
+
+    /// <summary>Records that a per-site live-alarm aggregator stopped (decrements the active-aggregator gauge).</summary>
+    public static void LiveAlarmAggregatorStopped() => Interlocked.Decrement(ref _liveAlarmAggregatorsActive);
+
+    /// <summary>Records that a per-site live-alarm aggregator re-established its site-wide gRPC stream.</summary>
+    public static void RecordLiveAlarmStreamReconnect() => _liveAlarmStreamReconnects.Add(1);
+
     /// <summary>
     /// Registers the provider the StoreAndForward queue-depth gauge reads on each observation.
     /// A later task supplies a provider that reads the real StoreAndForward depth. A null

src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs

@@ -1,6 +1,7 @@
 using Akka.Actor;
 using Akka.Event;
 using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
+using ZB.MOM.WW.ScadaBridge.Commons.Observability;
 using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
 
 namespace ZB.MOM.WW.ScadaBridge.Communication.Actors;
@@ -182,6 +183,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
     protected override void PreStart()
     {
         _log.Info("Starting site-alarm aggregator for site {0}", _siteIdentifier);
+        // Telemetry: this aggregator is now a running per-site live cache (gauge +1). Balanced
+        // in PostStop, which Akka always runs on termination for any reason.
+        ScadaBridgeTelemetry.LiveAlarmAggregatorStarted();
         _lifetimeCts = new CancellationTokenSource();
 
         // Stream-first: open the site-wide alarm stream BEFORE the first seed so deltas
@@ -204,6 +208,8 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
         _lifetimeCts?.Cancel();
         _lifetimeCts?.Dispose();
         _lifetimeCts = null;
+        // Telemetry: this aggregator is no longer running (gauge -1). Balances PreStart.
+        ScadaBridgeTelemetry.LiveAlarmAggregatorStopped();
         base.PostStop();
     }
 
@@ -223,6 +229,8 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
         {
             _log.Info("Site-alarm gRPC stream for {0} was down; reopening on reconcile tick", _siteIdentifier);
             _retryCount = 0;
+            // Telemetry: a reconcile-driven reopen after the stream was given up is a reconnect.
+            ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();
             OpenGrpcStream();
         }
     }
@@ -455,6 +463,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
         // Flip to the other node.
         _useNodeA = !_useNodeA;
 
+        // Telemetry: a NodeA↔NodeB failover flip is a reconnect + re-seed.
+        ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();
+
         // A failover flip must RE-SEED (never silently serve stale) — kick a reconcile
         // fan-out alongside the reconnect. Buffering during the fan-out keeps the new
         // stream's deltas coherent with the fresh snapshot.

src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs

@@ -65,5 +65,25 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase<Communi
 
         builder.RequireThat(options.GrpcMaxConcurrentStreams > 0,
             $"Communication:GrpcMaxConcurrentStreams must be positive (was {options.GrpcMaxConcurrentStreams}).");
+
+        // ── Aggregated live alarm cache (plan #10, Task 6) ───────────────────────
+        // Linger drives a Timer dueTime; TimeSpan.Zero is valid (stop the aggregator
+        // immediately when the last viewer leaves), only a negative value is invalid.
+        builder.RequireThat(options.LiveAlarmCacheLinger >= TimeSpan.Zero,
+            $"Communication:LiveAlarmCacheLinger must be zero or a positive duration (was {options.LiveAlarmCacheLinger}).");
+
+        // Reconcile interval is a periodic timer cadence AND the start-retry cadence; a
+        // zero/negative value would spin or never fire.
+        builder.RequireThat(options.LiveAlarmCacheReconcileInterval > TimeSpan.Zero,
+            $"Communication:LiveAlarmCacheReconcileInterval must be a positive duration (was {options.LiveAlarmCacheReconcileInterval}).");
+
+        // Seed fan-out concurrency gates a SemaphoreSlim; at least 1, capped so a
+        // fat-fingered config can't open an unbounded burst of cross-cluster Asks.
+        builder.RequireThat(options.LiveAlarmCacheSeedConcurrency is >= 1 and <= 64,
+            $"Communication:LiveAlarmCacheSeedConcurrency must be between 1 and 64 (was {options.LiveAlarmCacheSeedConcurrency}).");
+
+        // Per-site viewer cap must admit at least one viewer, else the page could never go live.
+        builder.RequireThat(options.LiveAlarmCacheMaxSubscribersPerSite >= 1,
+            $"Communication:LiveAlarmCacheMaxSubscribersPerSite must be at least 1 (was {options.LiveAlarmCacheMaxSubscribersPerSite}).");
     }
 }

src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs

@@ -92,12 +92,18 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache
             // A pending linger stop is now moot — a viewer arrived. Invalidate it.
             entry.CancelLinger();
 
+            // Enforce the per-site viewer cap — fail SAFE: reject the excess viewer with a
+            // no-op disposable rather than throwing (this runs on a Blazor render path) or
+            // growing the subscriber list unbounded. The shared aggregator keeps serving the
+            // already-registered viewers; the rejected circuit just keeps its 15s poll
+            // fallback. Reaching the cap almost always means a Dispose leak or a runaway page.
             if (entry.Subscribers.Count >= _options.LiveAlarmCacheMaxSubscribersPerSite)
             {
                 _logger.LogWarning(
-                    "Site {SiteId} live alarm cache already has {Count} viewers (cap {Cap}); " +
-                    "new viewer still registered but this indicates a leak or a very busy page",
-                    siteId, entry.Subscribers.Count, _options.LiveAlarmCacheMaxSubscribersPerSite);
+                    "Site {SiteId} live alarm cache is at its viewer cap ({Cap}); rejecting the new " +
+                    "viewer (it will keep polling). This indicates a subscription leak or a very busy page.",
+                    siteId, _options.LiveAlarmCacheMaxSubscribersPerSite);
+                return NoOpSubscription.Instance;
             }
 
             entry.Subscribers.Add(subscription);
@@ -446,6 +452,18 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache
         }
     }
 
+    /// <summary>
+    /// The handle returned when a Subscribe is rejected at the per-site viewer cap. It was
+    /// never registered, so <see cref="Dispose"/> is a genuine no-op — safe to dispose any
+    /// number of times and it never touches the site's subscriber list.
+    /// </summary>
+    private sealed class NoOpSubscription : IDisposable
+    {
+        public static readonly NoOpSubscription Instance = new();
+        private NoOpSubscription() { }
+        public void Dispose() { }
+    }
+
     /// <summary>The disposable handed to a viewer; idempotent <see cref="Dispose"/> unregisters it.</summary>
     private sealed class Subscription : IDisposable
     {

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

@@ -36,4 +36,54 @@ public class CommunicationOptionsValidatorTests
         Assert.True(result.Failed);
         Assert.Contains("GrpcMaxConcurrentStreams", result.FailureMessage);
     }
+
+    // ── Aggregated live alarm cache options (plan #10, Task 6) ───────────────────
+
+    [Fact]
+    public void ZeroLiveAlarmCacheLinger_IsValid()
+    {
+        // Zero linger = stop the aggregator immediately when the last viewer leaves.
+        var result = Validate(new CommunicationOptions { LiveAlarmCacheLinger = TimeSpan.Zero });
+        Assert.True(result.Succeeded, result.FailureMessage);
+    }
+
+    [Fact]
+    public void NegativeLiveAlarmCacheLinger_IsRejected()
+    {
+        var result = Validate(new CommunicationOptions { LiveAlarmCacheLinger = TimeSpan.FromSeconds(-1) });
+        Assert.True(result.Failed);
+        Assert.Contains("LiveAlarmCacheLinger", result.FailureMessage);
+    }
+
+    [Fact]
+    public void ZeroLiveAlarmCacheReconcileInterval_IsRejected()
+    {
+        var result = Validate(new CommunicationOptions { LiveAlarmCacheReconcileInterval = TimeSpan.Zero });
+        Assert.True(result.Failed);
+        Assert.Contains("LiveAlarmCacheReconcileInterval", result.FailureMessage);
+    }
+
+    [Fact]
+    public void ZeroLiveAlarmCacheSeedConcurrency_IsRejected()
+    {
+        var result = Validate(new CommunicationOptions { LiveAlarmCacheSeedConcurrency = 0 });
+        Assert.True(result.Failed);
+        Assert.Contains("LiveAlarmCacheSeedConcurrency", result.FailureMessage);
+    }
+
+    [Fact]
+    public void ExcessiveLiveAlarmCacheSeedConcurrency_IsRejected()
+    {
+        var result = Validate(new CommunicationOptions { LiveAlarmCacheSeedConcurrency = 65 });
+        Assert.True(result.Failed);
+        Assert.Contains("LiveAlarmCacheSeedConcurrency", result.FailureMessage);
+    }
+
+    [Fact]
+    public void ZeroLiveAlarmCacheMaxSubscribersPerSite_IsRejected()
+    {
+        var result = Validate(new CommunicationOptions { LiveAlarmCacheMaxSubscribersPerSite = 0 });
+        Assert.True(result.Failed);
+        Assert.Contains("LiveAlarmCacheMaxSubscribersPerSite", result.FailureMessage);
+    }
 }

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

@@ -54,7 +54,8 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
         public override SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint) => _client;
     }
 
-    private SiteAlarmLiveCacheService CreateService(TimeSpan linger, out CountingFactory factory)
+    private SiteAlarmLiveCacheService CreateService(TimeSpan linger, out CountingFactory factory,
+        int maxSubscribersPerSite = 200)
     {
         // Site with gRPC addresses, and NO enabled instances → the seed fan-out returns
         // empty immediately (so IsLive flips true fast without any snapshot Asks).
@@ -80,7 +81,8 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
         var options = Options.Create(new CommunicationOptions
         {
             LiveAlarmCacheLinger = linger,
-            LiveAlarmCacheReconcileInterval = TimeSpan.FromMinutes(10)
+            LiveAlarmCacheReconcileInterval = TimeSpan.FromMinutes(10),
+            LiveAlarmCacheMaxSubscribersPerSite = maxSubscribersPerSite
         });
 
         var comm = new CommunicationService(Options.Create(new CommunicationOptions()),
@@ -214,6 +216,38 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
         AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5));
     }
 
+    [Fact]
+    public void Viewer_Over_The_Per_Site_Cap_Is_Rejected_Safely_And_Not_Counted()
+    {
+        // Cap of 2: the first two viewers register; a third is rejected fail-safe with a
+        // no-op handle (never throws into the render path, never grows the list). Proof it
+        // was NOT counted: disposing only the two real viewers tears the aggregator down
+        // after the linger — if the rejected viewer had been registered, one subscriber
+        // would remain and the site would stay live forever.
+        var service = CreateService(TimeSpan.FromMilliseconds(200), out _, maxSubscribersPerSite: 2);
+
+        var sub1 = service.Subscribe(SiteId, () => { });
+        var sub2 = service.Subscribe(SiteId, () => { });
+        AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5));
+
+        // Third viewer is over the cap.
+        var overflow = service.Subscribe(SiteId, () => { });
+        Assert.NotNull(overflow);
+        // Disposing the rejected handle is a genuine no-op — safe, idempotent, touches nothing.
+        overflow.Dispose();
+        overflow.Dispose();
+
+        // Still live with the two real viewers present.
+        Assert.True(service.IsLive(SiteId));
+
+        sub1.Dispose();
+        sub2.Dispose();
+
+        // Both real viewers gone → after the linger the aggregator stops. This only holds
+        // if the overflow viewer was never added to the subscriber list.
+        AwaitCondition(() => !service.IsLive(SiteId), TimeSpan.FromSeconds(3));
+    }
+
     [Fact]
     public void Unknown_Site_Registers_Viewer_But_Never_Goes_Live()
     {

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

@@ -0,0 +1,94 @@
+using System.Diagnostics.Metrics;
+using ZB.MOM.WW.ScadaBridge.Commons.Observability;
+
+namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
+
+/// <summary>
+/// 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
+/// relative to a baseline so the process-wide static instruments are robust to any
+/// parallel test interleaving.
+/// </summary>
+public class SiteAlarmLiveCacheTelemetryTests
+{
+    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;
+    }
+
+    private static long ReadCounter(string instrumentName, Action act)
+    {
+        long delta = 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, _, _) => delta += measurement);
+        listener.Start();
+        act();
+        return delta;
+    }
+
+    [Fact]
+    public void ActiveAggregatorGauge_ReflectsStartAndStop()
+    {
+        const string gauge = "scadabridge.site.alarm_cache.aggregators.active";
+        var baseline = ReadGauge(gauge);
+
+        ScadaBridgeTelemetry.LiveAlarmAggregatorStarted();
+        try
+        {
+            Assert.Equal(baseline + 1, ReadGauge(gauge));
+
+            ScadaBridgeTelemetry.LiveAlarmAggregatorStarted();
+            try
+            {
+                Assert.Equal(baseline + 2, ReadGauge(gauge));
+            }
+            finally
+            {
+                ScadaBridgeTelemetry.LiveAlarmAggregatorStopped();
+            }
+
+            Assert.Equal(baseline + 1, ReadGauge(gauge));
+        }
+        finally
+        {
+            ScadaBridgeTelemetry.LiveAlarmAggregatorStopped();
+        }
+
+        Assert.Equal(baseline, ReadGauge(gauge));
+    }
+
+    [Fact]
+    public void ReconnectCounter_IncrementsOnEachRecordedReconnect()
+    {
+        var delta = ReadCounter("scadabridge.site.alarm_cache.reconnects", () =>
+        {
+            ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();
+            ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();
+        });
+
+        Assert.Equal(2, delta);
+    }
+}

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

@@ -24,6 +24,9 @@ public class ScadaBridgeTelemetryTests
             ScadaBridgeTelemetry.RecordInboundApiRequest("X");
             ScadaBridgeTelemetry.SiteConnectionOpened();
             ScadaBridgeTelemetry.SiteConnectionClosed();
+            ScadaBridgeTelemetry.LiveAlarmAggregatorStarted();
+            ScadaBridgeTelemetry.LiveAlarmAggregatorStopped();
+            ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();
             ScadaBridgeTelemetry.SetQueueDepthProvider(() => 5);
         });