From c4f97d0e87a13a4efa5e225c47d482eea29910e4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 10 Jul 2026 12:33:40 -0400 Subject: [PATCH] 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 --- .../Observability/ScadaBridgeTelemetry.cs | 28 ++++++ .../Actors/SiteAlarmAggregatorActor.cs | 11 +++ .../CommunicationOptionsValidator.cs | 20 ++++ .../SiteAlarmLiveCacheService.cs | 24 ++++- .../CommunicationOptionsValidatorTests.cs | 50 ++++++++++ .../SiteAlarmLiveCacheServiceTests.cs | 38 +++++++- .../SiteAlarmLiveCacheTelemetryTests.cs | 94 +++++++++++++++++++ .../ScadaBridgeTelemetryTests.cs | 3 + 8 files changed, 263 insertions(+), 5 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index d6b5bc54..1f8cc2ad 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -47,11 +47,23 @@ public static class ScadaBridgeTelemetry Meter.CreateCounter("scadabridge.store_and_forward.replication.failures", unit: "1", description: "S&F buffer replication operations that failed to dispatch/deliver to the peer node"); + /// + /// 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. + /// + private static readonly Counter _liveAlarmStreamReconnects = + Meter.CreateCounter("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 ---------------- /// Current count of open site connections, mutated via . private static long _siteConnectionsUp; + /// Current count of running per-site live-alarm aggregators, mutated via . + private static long _liveAlarmAggregatorsActive; + /// Provider that yields the live StoreAndForward queue depth; set by a later task. private static Func? _queueDepthProvider; @@ -68,6 +80,13 @@ public static class ScadaBridgeTelemetry () => Volatile.Read(ref _queueDepthProvider) is { } p ? p() : 0L, unit: "items", description: "Current StoreAndForward queue depth."); + + /// Gauge reporting the number of currently running per-site live-alarm aggregators. + private static readonly ObservableGauge _liveAlarmAggregatorsActiveGauge = + Meter.CreateObservableGauge("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 /// Records that a site connection closed (decrements the up-count gauge). public static void SiteConnectionClosed() => Interlocked.Decrement(ref _siteConnectionsUp); + /// Records that a per-site live-alarm aggregator started (increments the active-aggregator gauge). + public static void LiveAlarmAggregatorStarted() => Interlocked.Increment(ref _liveAlarmAggregatorsActive); + + /// Records that a per-site live-alarm aggregator stopped (decrements the active-aggregator gauge). + public static void LiveAlarmAggregatorStopped() => Interlocked.Decrement(ref _liveAlarmAggregatorsActive); + + /// Records that a per-site live-alarm aggregator re-established its site-wide gRPC stream. + public static void RecordLiveAlarmStreamReconnect() => _liveAlarmStreamReconnects.Add(1); + /// /// 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 diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs index 951f1174..7d19e0cb 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs +++ b/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. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs index d101741c..19236555 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs @@ -65,5 +65,25 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase 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})."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs index c698dcc9..ea0c5036 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs +++ b/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 } } + /// + /// The handle returned when a Subscribe is rejected at the per-site viewer cap. It was + /// never registered, so is a genuine no-op — safe to dispose any + /// number of times and it never touches the site's subscriber list. + /// + private sealed class NoOpSubscription : IDisposable + { + public static readonly NoOpSubscription Instance = new(); + private NoOpSubscription() { } + public void Dispose() { } + } + /// The disposable handed to a viewer; idempotent unregisters it. private sealed class Subscription : IDisposable { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs index 2ce5f0ff..c29d677c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs +++ b/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); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs index 40d56b68..3d7e07ad 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs +++ b/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() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs new file mode 100644 index 00000000..87549372 --- /dev/null +++ b/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; + +/// +/// 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 SiteStreamGrpcServerTests.SiteConnectionUpGauge_*; reads are +/// relative to a baseline so the process-wide static instruments are robust to any +/// parallel test interleaving. +/// +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((_, 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((_, 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); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs index c081386e..6c89f559 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs +++ b/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); });