diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs index 7d19e0cb..98cceee7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs @@ -52,11 +52,16 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers private readonly string _grpcNodeAAddress; private readonly string _grpcNodeBAddress; private readonly TimeSpan _reconcileInterval; + private readonly TimeSpan _publishCoalesce; private const int MaxRetries = 3; private const string ReconnectTimerKey = "alarm-grpc-reconnect"; private const string StabilityTimerKey = "alarm-grpc-stability"; private const string ReconcileTimerKey = "alarm-reconcile"; + private const string PublishTimerKey = "alarm-publish-coalesce"; + + /// True while a coalesced publish is armed (dirty deltas awaiting one tick). Actor-thread only. + private bool _publishPending; /// Delay between gRPC reconnection attempts. Settable for tests. internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5); @@ -118,6 +123,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers /// gRPC address of the site's node A. /// gRPC address of the site's node B. /// Periodic reconcile snapshot cadence. + /// + /// Publish-coalescing window for live deltas: a positive value batches a delta storm + /// into one publish per window (review 02 round 2, N6); + /// restores per-delta publishing (legacy). Seed/reconcile publishes stay immediate. + /// public SiteAlarmAggregatorActor( string siteIdentifier, string correlationId, @@ -126,7 +136,8 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers SiteStreamGrpcClientFactory grpcFactory, string grpcNodeAAddress, string grpcNodeBAddress, - TimeSpan reconcileInterval) + TimeSpan reconcileInterval, + TimeSpan publishCoalesce) { _siteIdentifier = siteIdentifier; _correlationId = correlationId; @@ -136,6 +147,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _grpcNodeAAddress = grpcNodeAAddress; _grpcNodeBAddress = grpcNodeBAddress; _reconcileInterval = reconcileInterval; + _publishCoalesce = publishCoalesce; // Live delta from the site-wide alarm stream (marshalled in via Self.Tell). // A received delta must NOT reset the retry budget (a flapping stream that @@ -152,6 +164,13 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Periodic reconcile tick (and the re-seed kicked after a reconnect). Receive(_ => OnReconcileTick()); + // Coalesced-publish tick: one publish for a batch of dirtying deltas (N6). + Receive(_ => + { + _publishPending = false; + if (!_stopped) Publish(); + }); + // Stream stayed up for StabilityWindow — recover the retry budget. Receive(_ => { @@ -300,6 +319,10 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _log.Debug("Site-alarm {0} {1} complete: {2} alarm row(s)", _siteIdentifier, msg.IsInitial ? "seed" : "reconcile", _cache.Count); + // The fresh snapshot already carries the buffered deltas; drop any armed coalesce + // tick so we publish once, immediately. + Timers.Cancel(PublishTimerKey); + _publishPending = false; Publish(); } @@ -319,7 +342,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Only publish if we already had a seed (so IsLive doesn't flip true on a // failed initial seed — the page keeps its poll fallback until we truly seed). if (_seeded) + { + Timers.Cancel(PublishTimerKey); + _publishPending = false; Publish(); + } } /// @@ -362,9 +389,23 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers return; } - // Pass-through: apply and publish only if the cache actually changed. + // Pass-through: apply and (coalesced) publish only if the cache actually changed. if (ApplyDelta(delta, requireStrictlyNewer: false)) - Publish(); + SchedulePublish(); + } + + /// + /// Coalesced publish: with a positive window, the first dirtying delta arms a + /// single-shot timer and further deltas ride the same tick — one snapshot copy and + /// one viewer fan-out per window instead of per transition (N6). Zero = legacy + /// immediate publish. Last write wins, so batching never changes final state. + /// + private void SchedulePublish() + { + if (_publishCoalesce <= TimeSpan.Zero) { Publish(); return; } + if (_publishPending) return; + _publishPending = true; + Timers.StartSingleTimer(PublishTimerKey, new PublishCoalesced(), _publishCoalesce); } /// @@ -520,6 +561,9 @@ internal sealed record SeedFailed(Exception Exception, bool IsInitial); /// Internal: periodic reconcile tick (and the re-seed kicked after a reconnect). internal sealed record RunReconcile; +/// Internal: coalesced-publish tick — flush the dirty cache to viewers once (N6). +internal sealed record PublishCoalesced; + /// Internal: site-alarm gRPC stream error occurred. internal sealed record GrpcAlarmStreamError(Exception Exception); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs index c5957947..fc9e9ffe 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs @@ -122,4 +122,13 @@ public class CommunicationOptions /// count; this just bounds the subscriber list. Default 200. /// public int LiveAlarmCacheMaxSubscribersPerSite { get; set; } = 200; + + /// + /// Publish-coalescing window for live alarm deltas: an applied delta marks the cache + /// dirty and one publish (fresh snapshot + per-viewer onChanged fan-out) fires after + /// this window, batching an alarm storm into ~4 publishes/second instead of one per + /// transition (review 02 round 2, N6). Zero = publish per delta (legacy). Seed and + /// reconcile publishes are always immediate. Default 250 ms. + /// + public TimeSpan LiveAlarmCachePublishCoalesce { get; set; } = TimeSpan.FromMilliseconds(250); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs index 19236555..31151b67 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs @@ -85,5 +85,10 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase= 1, $"Communication:LiveAlarmCacheMaxSubscribersPerSite must be at least 1 (was {options.LiveAlarmCacheMaxSubscribersPerSite})."); + + // Publish-coalescing window drives a single-shot timer; TimeSpan.Zero is valid + // (publish per delta — legacy), only a negative value is invalid. + builder.RequireThat(options.LiveAlarmCachePublishCoalesce >= TimeSpan.Zero, + $"Communication:LiveAlarmCachePublishCoalesce must be zero or a positive duration (was {options.LiveAlarmCachePublishCoalesce})."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs index ea0c5036..568dc538 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs @@ -233,7 +233,8 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache _grpcFactory, grpcA, grpcB, - _options.LiveAlarmCacheReconcileInterval)); + _options.LiveAlarmCacheReconcileInterval, + _options.LiveAlarmCachePublishCoalesce)); entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}"); entry.StartRetryTimer?.Dispose(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs index c29d677c..dafba565 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs @@ -86,4 +86,22 @@ public class CommunicationOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("LiveAlarmCacheMaxSubscribersPerSite", result.FailureMessage); } + + // ── R2 T10: live-delta publish-coalescing window (N6) ──────────────────────── + + [Fact] + public void ZeroLiveAlarmCachePublishCoalesce_IsValid() + { + // Zero = publish per delta (legacy behavior). + var result = Validate(new CommunicationOptions { LiveAlarmCachePublishCoalesce = TimeSpan.Zero }); + Assert.True(result.Succeeded, result.FailureMessage); + } + + [Fact] + public void NegativeLiveAlarmCachePublishCoalesce_IsRejected() + { + var result = Validate(new CommunicationOptions { LiveAlarmCachePublishCoalesce = TimeSpan.FromMilliseconds(-1) }); + Assert.True(result.Failed); + Assert.Contains("LiveAlarmCachePublishCoalesce", result.FailureMessage); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs index 32846846..af3be444 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -137,7 +137,7 @@ public class SiteAlarmAggregatorActorTests : TestKit } private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor( - TimeSpan? reconcileInterval = null) + TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null) { var seed = new SeedStub(); var sink = new PublishSink(); @@ -145,7 +145,8 @@ public class SiteAlarmAggregatorActorTests : TestKit var props = Props.Create(() => new SiteAlarmAggregatorActor( SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB, - reconcileInterval ?? TimeSpan.FromMinutes(10))); + reconcileInterval ?? TimeSpan.FromMinutes(10), + publishCoalesce ?? TimeSpan.Zero)); var actor = Sys.ActorOf(props); return (actor, seed, sink, factory); @@ -396,6 +397,45 @@ public class SiteAlarmAggregatorActorTests : TestKit AwaitCondition(() => TotalSubs() > before, TimeSpan.FromSeconds(3)); } + // ── R2 T10: live-delta publish coalescing (N6) ── + + [Fact] + public void DeltaBurst_IsCoalesced_IntoFewPublishes_ContainingEveryRow() + { + var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.FromMilliseconds(150)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); // empty initial seed + AwaitAssert(() => Assert.Equal(1, sink.Count)); // seed publish, immediate + + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + var sub = factory.ClientFor(GrpcNodeA).Subs.Single(); + var now = DateTimeOffset.UtcNow; + for (var i = 0; i < 50; i++) + sub.OnAlarm(Alarm($"A{i}", "", 900, now)); // 50 distinct keys, one burst + + AwaitAssert(() => + { + Assert.Equal(50, sink.Latest!.Count); // nothing lost + Assert.InRange(sink.Count, 2, 4); // pre-fix: 51 publishes (1 seed + 1 per delta) + }, TimeSpan.FromSeconds(3)); + } + + [Fact] + public void ZeroCoalesce_PreservesLegacyPerDeltaPublish() + { + var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + AwaitAssert(() => Assert.Equal(1, sink.Count)); + + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + var sub = factory.ClientFor(GrpcNodeA).Subs.Single(); + var now = DateTimeOffset.UtcNow; + sub.OnAlarm(Alarm("A1", "", 900, now)); + sub.OnAlarm(Alarm("A2", "", 900, now)); + AwaitAssert(() => Assert.Equal(3, sink.Count)); // 1 seed + 1 per delta + } + [Fact] public void Stop_Message_TearsDown_Grpc_And_Stops_Actor() {