diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs new file mode 100644 index 00000000..951f1174 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs @@ -0,0 +1,519 @@ +using Akka.Actor; +using Akka.Event; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Actors; + +/// +/// One-per-site aggregator on the active central node backing the operator Alarm +/// Summary live cache (plan #10, Task 4). Holds a transient in-memory +/// Dictionary<AlarmKey, AlarmStateChanged> of the whole site's current +/// alarm state — NO persistence (locked [PERM]: no central alarm store). Created +/// and torn down by under viewer reference-count. +/// +/// Seed-then-stream ordering (copied from ): +/// the site-wide, alarm-only gRPC stream (SubscribeSite) is opened FIRST in +/// so live deltas start flowing during the seed's snapshot-build +/// + transit window; deltas arriving while a seed/reconcile fan-out is in flight are +/// buffered in arrival order. When the fan-out completes the cache is rebuilt +/// authoritatively from the fresh snapshot, then the buffer is flushed with per-key +/// dedup against the seed, then the actor passes live deltas straight into the cache. +/// +/// +/// Placeholder reconciliation: the snapshot fan-out carries +/// rows (a configured native +/// source binding with no active conditions); the live stream drops them server-side. +/// Placeholders are seeded and never expected on the live stream — a live delta for a +/// real alarm has a distinct AlarmKey so it can never wipe a placeholder row and +/// vice-versa. Placeholder-vs-real coherence is refreshed by the periodic reconcile. +/// +/// +/// Failover + drift: a gRPC error flips NodeA↔NodeB with the same retry budget + +/// stability window as , and each reconnect triggers +/// a RE-SEED (never silently serve stale). A periodic reconcile snapshot +/// (, default 60s) corrects instance-set drift and any +/// missed delta. +/// +/// All state is mutated only on the actor thread: gRPC callbacks and fan-out results are +/// marshalled back via Self.Tell, so the cache needs no internal lock. The +/// published snapshot handed to the service is a fresh immutable list (reference swap), +/// so Blazor render threads never observe a partially-mutated cache. +/// +public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers +{ + private readonly ILoggingAdapter _log = Context.GetLogger(); + private readonly string _siteIdentifier; + private readonly string _correlationId; + private readonly Func>> _seedFn; + private readonly Action> _publish; + private readonly SiteStreamGrpcClientFactory _grpcFactory; + private readonly string _grpcNodeAAddress; + private readonly string _grpcNodeBAddress; + private readonly TimeSpan _reconcileInterval; + + 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"; + + /// Delay between gRPC reconnection attempts. Settable for tests. + internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// How long a freshly-opened gRPC stream must stay up before its retry budget is + /// considered recovered (mirrors ). + /// Settable for tests. + /// + internal static TimeSpan StabilityWindow { get; set; } = TimeSpan.FromSeconds(60); + + private int _retryCount; + private bool _useNodeA = true; + private bool _stopped; + + /// + /// True once the live gRPC stream has been given up (retry budget exhausted). Reconcile + /// snapshots keep serving in the meantime; the next reconcile tick self-heals the stream + /// by resetting the retry budget and reopening it, so a sustained site outage does not + /// permanently drop the live feed. Actor-thread only. + /// + private bool _streamDown; + private CancellationTokenSource? _grpcCts; + private CancellationTokenSource? _lifetimeCts; + + /// Current whole-site alarm state, keyed by . Actor-thread only. + private readonly Dictionary _cache = new(); + + /// True once the first seed has completed and been published. Actor-thread only. + private bool _seeded; + + /// True while a seed/reconcile snapshot fan-out is in flight (deltas buffer). Actor-thread only. + private bool _fanoutInFlight; + + /// Ordered buffer of live deltas that arrived while a fan-out was in flight. Actor-thread only. + private readonly List _buffer = new(); + + private const int BufferWarnThreshold = 10_000; + private bool _bufferWarned; + + /// + public ITimerScheduler Timers { get; set; } = null!; + + /// + /// Creates a per-site alarm aggregator. + /// + /// Site identifier (for logging / gRPC client keying). + /// Correlation id for the site-wide gRPC subscription. + /// + /// Snapshot fan-out that returns the whole site's current alarm rows (including + /// placeholders), best-effort and tolerant of per-instance failure. Re-run on every + /// seed and reconcile, so it re-enumerates the site's Enabled instances each call. + /// + /// + /// Publishes a fresh immutable snapshot of the cache to the owning service, which + /// stores it and raises viewer onChanged callbacks. Invoked on the actor thread. + /// + /// Factory caching one gRPC client per (site, endpoint). + /// gRPC address of the site's node A. + /// gRPC address of the site's node B. + /// Periodic reconcile snapshot cadence. + public SiteAlarmAggregatorActor( + string siteIdentifier, + string correlationId, + Func>> seedFn, + Action> publish, + SiteStreamGrpcClientFactory grpcFactory, + string grpcNodeAAddress, + string grpcNodeBAddress, + TimeSpan reconcileInterval) + { + _siteIdentifier = siteIdentifier; + _correlationId = correlationId; + _seedFn = seedFn; + _publish = publish; + _grpcFactory = grpcFactory; + _grpcNodeAAddress = grpcNodeAAddress; + _grpcNodeBAddress = grpcNodeBAddress; + _reconcileInterval = reconcileInterval; + + // 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 + // delivers one delta between failures would otherwise never trip MaxRetries). + Receive(HandleLiveDelta); + + // A seed/reconcile fan-out completed. + Receive(OnSeedCompleted); + + // A seed/reconcile fan-out threw as a whole (individual per-instance faults are + // swallowed inside seedFn and degrade to fewer rows, not a whole-fan-out failure). + Receive(OnSeedFailed); + + // Periodic reconcile tick (and the re-seed kicked after a reconnect). + Receive(_ => OnReconcileTick()); + + // Stream stayed up for StabilityWindow — recover the retry budget. + Receive(_ => + { + if (_stopped) return; + _retryCount = 0; + _log.Debug("Site-alarm gRPC stream for {0} stable; retry count reset", _siteIdentifier); + }); + + // gRPC stream error — flip node + reconnect + re-seed. + Receive(msg => + { + _log.Warning("Site-alarm gRPC stream error for {0}: {1}", _siteIdentifier, msg.Exception.Message); + HandleGrpcError(); + }); + + Receive(_ => OpenGrpcStream()); + + // Owning service asks us to stop (last viewer left + linger elapsed). + Receive(_ => + { + _log.Info("Stopping site-alarm aggregator for {0}", _siteIdentifier); + CleanupGrpc(); + _stopped = true; + Context.Stop(Self); + }); + } + + /// + protected override void PreStart() + { + _log.Info("Starting site-alarm aggregator for site {0}", _siteIdentifier); + _lifetimeCts = new CancellationTokenSource(); + + // Stream-first: open the site-wide alarm stream BEFORE the first seed so deltas + // in the seed window are captured (buffered) rather than lost. + OpenGrpcStream(); + + // Kick the initial seed fan-out. + StartFanout(isInitial: true); + + // Periodic reconcile backstop. + Timers.StartPeriodicTimer(ReconcileTimerKey, new RunReconcile(), _reconcileInterval, _reconcileInterval); + } + + /// + protected override void PostStop() + { + _grpcCts?.Cancel(); + _grpcCts?.Dispose(); + _grpcCts = null; + _lifetimeCts?.Cancel(); + _lifetimeCts?.Dispose(); + _lifetimeCts = null; + base.PostStop(); + } + + // ── Reconcile tick ────────────────────────────────────────────────────────── + + /// + /// Periodic reconcile: always re-run the snapshot fan-out (corrects drift + missed + /// deltas), and if the live stream was previously given up, self-heal it by resetting + /// the retry budget and reopening — so a sustained outage never permanently kills the + /// live feed. + /// + private void OnReconcileTick() + { + if (_stopped) return; + StartFanout(isInitial: false); + if (_streamDown) + { + _log.Info("Site-alarm gRPC stream for {0} was down; reopening on reconcile tick", _siteIdentifier); + _retryCount = 0; + OpenGrpcStream(); + } + } + + // ── Seed / reconcile fan-out ──────────────────────────────────────────────── + + /// + /// Kicks a snapshot fan-out as a background task, marshalling the result back via + /// Self.Tell. While in flight, live deltas buffer. A reconcile that arrives + /// while a fan-out is already running is skipped (no stacking). + /// + private void StartFanout(bool isInitial) + { + if (_stopped) return; + if (_fanoutInFlight) + { + // Already seeding/reconciling — don't stack a second fan-out. + return; + } + + _fanoutInFlight = true; + var self = Self; + var ct = _lifetimeCts?.Token ?? CancellationToken.None; + + Task.Run(async () => + { + try + { + var alarms = await _seedFn(ct); + self.Tell(new SeedCompleted(alarms, isInitial)); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + // Actor stopping — drop silently. + } + catch (Exception ex) + { + self.Tell(new SeedFailed(ex, isInitial)); + } + }, ct); + } + + private void OnSeedCompleted(SeedCompleted msg) + { + if (_stopped) return; + + // Rebuild the cache authoritatively from the fresh snapshot (this is what makes + // reconcile able to DROP rows for instances/alarms that disappeared — a merge + // could never remove a stale row since the live stream sends no "removed" event). + _cache.Clear(); + foreach (var alarm in msg.Alarms) + { + var key = AlarmKey(alarm); + if (!_cache.TryGetValue(key, out var existing) || alarm.Timestamp >= existing.Timestamp) + _cache[key] = alarm; + } + + // Flush the deltas buffered during the fan-out, deduped against the seed: a + // buffered delta whose key is in the fresh snapshot with an equal-or-newer + // timestamp is already reflected → drop; a strictly-newer (or new-key) delta is + // applied. Inclusive-on-snapshot boundary matches DebugStreamBridgeActor. + FlushBuffer(); + + _fanoutInFlight = false; + _seeded = true; + + _log.Debug("Site-alarm {0} {1} complete: {2} alarm row(s)", + _siteIdentifier, msg.IsInitial ? "seed" : "reconcile", _cache.Count); + + Publish(); + } + + private void OnSeedFailed(SeedFailed msg) + { + if (_stopped) return; + + _log.Warning(msg.Exception, + "Site-alarm {0} {1} fan-out failed; keeping current cache and relying on the next reconcile", + _siteIdentifier, msg.IsInitial ? "seed" : "reconcile"); + + // Don't lose deltas captured during the failed window — apply them pass-through + // into the (possibly stale/empty) cache. The next reconcile re-seeds authoritatively. + _fanoutInFlight = false; + FlushBuffer(dedupAgainstSeed: false); + + // 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) + Publish(); + } + + /// + /// Flushes the pre-fan-out buffer in arrival order. When + /// is true (normal completion) a buffered delta already reflected in the just-installed + /// snapshot (same key, timestamp <= cache entry) is dropped; otherwise every buffered + /// delta is applied pass-through. + /// + private void FlushBuffer(bool dedupAgainstSeed = true) + { + if (_buffer.Count == 0) return; + + foreach (var delta in _buffer) + { + if (dedupAgainstSeed) + ApplyDelta(delta, requireStrictlyNewer: true); + else + ApplyDelta(delta, requireStrictlyNewer: false); + } + _buffer.Clear(); + } + + // ── Live delta handling ───────────────────────────────────────────────────── + + private void HandleLiveDelta(AlarmStateChanged delta) + { + if (_stopped) return; + + if (_fanoutInFlight) + { + _buffer.Add(delta); + if (!_bufferWarned && _buffer.Count > BufferWarnThreshold) + { + _bufferWarned = true; + _log.Warning( + "Site-alarm pre-seed buffer for {0} exceeded {1} deltas while a fan-out was in flight " + + "(deltas retained, not dropped).", + _siteIdentifier, BufferWarnThreshold); + } + return; + } + + // Pass-through: apply and publish only if the cache actually changed. + if (ApplyDelta(delta, requireStrictlyNewer: false)) + Publish(); + } + + /// + /// Applies one alarm delta to the cache keyed by . Returns + /// true if the cache changed. A stale (older) delta for an existing key is + /// ignored. Never carries a placeholder (the live stream drops those), so a live + /// delta for a real alarm can only add/replace its own key — it never touches a + /// placeholder row under a different key. + /// + private bool ApplyDelta(AlarmStateChanged delta, bool requireStrictlyNewer) + { + var key = AlarmKey(delta); + if (_cache.TryGetValue(key, out var existing)) + { + var newer = requireStrictlyNewer + ? delta.Timestamp > existing.Timestamp + : delta.Timestamp >= existing.Timestamp; + if (!newer) return false; + } + _cache[key] = delta; + return true; + } + + private void Publish() + { + // Fresh immutable list — reference swap; readers never see a partial mutation. + var snapshot = _cache.Values.ToList(); + try + { + _publish(snapshot); + } + catch (Exception ex) + { + _log.Warning(ex, "Site-alarm publish callback threw for {0}; ignoring", _siteIdentifier); + } + } + + // ── gRPC stream lifecycle (mirrors DebugStreamBridgeActor) ────────────────── + + private void OpenGrpcStream() + { + if (_stopped) return; + + _streamDown = false; + var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + _log.Info("Opening site-alarm gRPC stream for {0} to {1}", _siteIdentifier, endpoint); + + _grpcCts?.Cancel(); + _grpcCts?.Dispose(); + _grpcCts = new CancellationTokenSource(); + + Timers.StartSingleTimer(StabilityTimerKey, new GrpcAlarmStreamStable(), StabilityWindow); + + var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint); + var self = Self; + var ct = _grpcCts.Token; + + Task.Run(async () => + { + await client.SubscribeSiteAsync( + _correlationId, + alarm => self.Tell(alarm), + ex => self.Tell(new GrpcAlarmStreamError(ex)), + ct); + }, ct); + } + + private void HandleGrpcError() + { + if (_stopped) return; + + // Stream failed before the stability window — retry budget NOT recovered. + Timers.Cancel(StabilityTimerKey); + + _retryCount++; + + if (_retryCount > MaxRetries) + { + // Give up the stream, but do NOT stop the aggregator: the periodic reconcile + // still refreshes the cache from ClusterClient snapshots, so the page keeps a + // (slower) live-ish view rather than going dark. A later reconcile-triggered + // reconnect is not attempted here; the stream is simply left down. + _log.Error("Site-alarm gRPC stream for {0} exceeded max retries ({1}); leaving stream down, " + + "reconcile snapshots continue and the next reconcile tick will retry the stream", + _siteIdentifier, MaxRetries); + _streamDown = true; + CleanupGrpc(); + return; + } + + // Unsubscribe the failed stream on the previous endpoint (TryGet, never + // GetOrCreate) so the site releases its relay actor instead of leaving a zombie. + var previousEndpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + _grpcFactory.TryGet(_siteIdentifier, previousEndpoint)?.Unsubscribe(_correlationId); + + // Flip to the other node. + _useNodeA = !_useNodeA; + + // 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. + StartFanout(isInitial: false); + + if (_retryCount == 1) + Self.Tell(new ReconnectAlarmStream()); + else + Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectAlarmStream(), ReconnectDelay); + } + + private void CleanupGrpc() + { + _grpcCts?.Cancel(); + _grpcCts?.Dispose(); + _grpcCts = null; + + var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + _grpcFactory.TryGet(_siteIdentifier, endpoint)?.Unsubscribe(_correlationId); + } + + // ── Dedup key (copied from DebugStreamBridgeActor.AlarmKey with attribution) ── + + /// + /// NUL delimiter so distinct identities never collide on a shared boundary. Cannot + /// appear in an instance/alarm name. Mirrors . + /// + private const char KeyDelimiter = '\u0000'; + + /// + /// Per-alarm dedup identity = (InstanceUniqueName, AlarmName, SourceReference) — + /// identical to DebugStreamBridgeActor.AlarmKey so native per-condition alarms + /// sharing an AlarmName but differing by source reference are not conflated. Each + /// nullable component is guarded to prevent silent null/empty key collisions. + /// + private static string AlarmKey(AlarmStateChanged a) => + string.Concat( + a.InstanceUniqueName ?? string.Empty, KeyDelimiter, + a.AlarmName ?? string.Empty, KeyDelimiter, + a.SourceReference ?? string.Empty); +} + +/// Message asking a to stop (last viewer left). +public sealed record StopSiteAlarmAggregator; + +/// Internal: a seed/reconcile snapshot fan-out completed with the whole-site alarm rows. +internal sealed record SeedCompleted(IReadOnlyList Alarms, bool IsInitial); + +/// Internal: a seed/reconcile snapshot fan-out threw as a whole. +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: site-alarm gRPC stream error occurred. +internal sealed record GrpcAlarmStreamError(Exception Exception); + +/// Internal: reconnect the site-alarm gRPC stream (flip node). +internal sealed record ReconnectAlarmStream; + +/// Internal: the current site-alarm gRPC stream has stayed up long enough to recover the retry budget. +internal sealed record GrpcAlarmStreamStable; diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs index 5ae541b9..c5957947 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs @@ -88,4 +88,38 @@ public class CommunicationOptions /// that are deployed once and never re-deployed. Default 1 hour ≫ . /// public TimeSpan PendingDeploymentPurgeInterval { get; set; } = TimeSpan.FromHours(1); + + // ── Aggregated live alarm cache (plan #10, Task 4) ─────────────────────────── + // Introduced by Task 4 with sane defaults; Task 6 formalizes eager validation + // (positive interval/linger, positive concurrency/subscriber cap) and telemetry. + + /// + /// How long the shared per-site live alarm aggregator keeps its gRPC stream + + /// in-memory cache warm after its LAST Alarm Summary viewer leaves, before it is + /// torn down. A short linger avoids re-seed thrash when an operator navigates away + /// and straight back. Default 30s. + /// + public TimeSpan LiveAlarmCacheLinger { get; set; } = TimeSpan.FromSeconds(30); + + /// + /// Cadence of the per-site aggregator's periodic reconcile snapshot fan-out. The + /// backstop that corrects instance-set drift (enable/disable/deploy/delete) and any + /// alarm delta missed by the live stream, since the aggregated stream never carries + /// an explicit "removed" event. Default 60s. + /// + public TimeSpan LiveAlarmCacheReconcileInterval { get; set; } = TimeSpan.FromSeconds(60); + + /// + /// Max concurrent per-instance snapshot fetches during a live-cache seed/reconcile + /// fan-out — the aggregated analogue of the Alarm Summary poll's + /// MaxConcurrentFetches. Default 8. + /// + public int LiveAlarmCacheSeedConcurrency { get; set; } = 8; + + /// + /// Upper bound on concurrent Alarm Summary viewers sharing one site's aggregator. + /// A defensive cap only — one gRPC stream per site is shared regardless of viewer + /// count; this just bounds the subscriber list. Default 200. + /// + public int LiveAlarmCacheMaxSubscribersPerSite { get; set; } = 200; } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs new file mode 100644 index 00000000..44a13efe --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs @@ -0,0 +1,62 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; + +namespace ZB.MOM.WW.ScadaBridge.Communication; + +/// +/// Central, transient, in-memory per-site live alarm cache (plan #10, Task 4). +/// A DI singleton shared by every Alarm Summary Blazor circuit on the active central +/// node. For each site with at least one active viewer it runs ONE shared per-site +/// aggregator that seeds from the existing debug-snapshot fan-out and stays warm on a +/// single site-wide, alarm-only gRPC stream (SubscribeSite), so the page +/// reflects alarm transitions in near-real-time instead of re-polling every 15s. +/// +/// Hard constraint (locked [PERM]): there is NO persisted central alarm +/// store. This cache is purely in-memory and lives only on the active central node — +/// no EF entity/table/migration backs it. On a NodeA↔NodeB failover the new active +/// node re-seeds from scratch. +/// +/// +/// Reference-counted lifecycle: the first for a site +/// starts its aggregator (open stream + snapshot seed); the last subscriber leaving +/// stops it after a short linger (see +/// ) to avoid re-seed thrash. +/// One gRPC stream per site, not per browser circuit. +/// +/// +public interface ISiteAlarmLiveCache +{ + /// + /// Registers a viewer for a site's live alarm feed and returns an + /// that unregisters it. The first subscriber for a site + /// starts the shared aggregator; the last subscriber's + /// schedules a linger-delayed stop. is raised (on the + /// aggregator's thread) after each applied change — the handler must not block and + /// should marshal any UI work onto its own dispatcher (mirrors + /// IDeploymentStatusNotifier). Disposing the returned handle is idempotent. + /// + /// The numeric site id whose alarms the viewer wants. + /// Callback invoked whenever the site's cached alarm set changes. + /// A disposable that unregisters this viewer (idempotent). + IDisposable Subscribe(int siteId, Action onChanged); + + /// + /// Returns the current in-memory alarm snapshot for a site — exactly what the page + /// should render. Empty when the site has no aggregator or has not yet completed its + /// first seed. Safe to call from Blazor render threads: it returns an immutable + /// point-in-time list (the aggregator publishes fresh immutable lists; readers never + /// see a partially-mutated collection). + /// + /// The numeric site id. + /// The current cached alarms (possibly empty), never null. + IReadOnlyList GetCurrentAlarms(int siteId); + + /// + /// Whether the site's aggregator has completed its initial seed and is serving live + /// state. The page can use this to decide whether to keep its 15s poll fallback + /// (Task 5): false means "not live yet — keep polling"; true means the + /// cache is authoritative. + /// + /// The numeric site id. + /// true once the site's aggregator has seeded and published at least once. + bool IsLive(int siteId); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/ServiceCollectionExtensions.cs index dc9577f2..0fb9ccdd 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ServiceCollectionExtensions.cs @@ -22,6 +22,13 @@ public static class ServiceCollectionExtensions services.AddSingleton(); services.AddSingleton(); + // Aggregated live alarm cache (plan #10, Task 4): transient, in-memory, shared + // per-site aggregator backing the operator Alarm Summary page. Singleton so every + // Blazor circuit shares one aggregator per site; started/stopped by viewer + // reference-count. No persistence (locked [PERM]: no central alarm store). + services.AddSingleton(); + services.AddSingleton(sp => sp.GetRequiredService()); + // Startup reconciliation handler — scoped (holds scoped repositories), resolved // per-request by CentralCommunicationActor inside a DI scope. Harmless on site // hosts: only the central actor ever resolves it. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs new file mode 100644 index 00000000..c698dcc9 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs @@ -0,0 +1,471 @@ +using System.Collections.Concurrent; +using Akka.Actor; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication; + +/// +/// Default — the DI singleton that owns the shared, +/// reference-counted, transient per-site live alarm cache (plan #10, Task 4). For each +/// site with active Alarm Summary viewers it runs ONE +/// , seeded by the existing debug-snapshot fan-out +/// () and kept warm on a +/// single site-wide, alarm-only gRPC stream. The actor holds the state and enforces +/// seed-then-stream ordering, dedup, placeholder coherence, failover re-seed and periodic +/// reconcile; this service is the DI façade: viewer reference-counting, a linger-delayed +/// last-out stop, the seed fan-out closure, and the published-snapshot store the page reads. +/// +/// No persistence (locked [PERM]): nothing here writes to the database — the +/// cache lives only in the actor's memory on the active central node. +/// +/// +public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache +{ + private static readonly IReadOnlyList Empty = Array.Empty(); + + private readonly IServiceProvider _serviceProvider; + private readonly CommunicationService _communicationService; + private readonly SiteStreamGrpcClientFactory _grpcFactory; + private readonly CommunicationOptions _options; + private readonly ILogger _logger; + + private readonly object _lock = new(); + private readonly Dictionary _sites = new(); + private ActorSystem? _actorSystem; + + /// + /// Initializes the live alarm cache service. + /// + /// Root provider; scopes are created per snapshot fan-out to resolve repositories. + /// Central-side comms used to Ask each instance for a debug snapshot (the seed). + /// Factory for the per-site site-wide alarm gRPC client. + /// Communication options (linger, reconcile interval, seed concurrency, subscriber cap). + /// Logger. + public SiteAlarmLiveCacheService( + IServiceProvider serviceProvider, + CommunicationService communicationService, + SiteStreamGrpcClientFactory grpcFactory, + IOptions options, + ILogger logger) + { + _serviceProvider = serviceProvider; + _communicationService = communicationService; + _grpcFactory = grpcFactory; + _options = options.Value; + _logger = logger; + } + + /// + /// Supplies the once it exists (called from + /// AkkaHostedService on the active central node, mirroring + /// ). Until set, + /// registers viewers but cannot start an aggregator, so the page keeps its poll fallback. + /// + /// The actor system to host per-site aggregators. + public void SetActorSystem(ActorSystem actorSystem) => _actorSystem = actorSystem; + + /// + public IDisposable Subscribe(int siteId, Action onChanged) + { + ArgumentNullException.ThrowIfNull(onChanged); + + var subscription = new Subscription(this, siteId, onChanged); + bool startNeeded; + SiteEntry entry; + + lock (_lock) + { + if (!_sites.TryGetValue(siteId, out entry!)) + { + entry = new SiteEntry(siteId); + _sites[siteId] = entry; + } + + // A pending linger stop is now moot — a viewer arrived. Invalidate it. + entry.CancelLinger(); + + 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); + } + + entry.Subscribers.Add(subscription); + + // Start the aggregator on the first viewer (or if a prior start failed and left + // no actor). Guarded by Starting so concurrent first-viewers don't double-start. + startNeeded = entry.Actor is null && !entry.Starting; + if (startNeeded) + entry.Starting = true; + } + + if (startNeeded) + _ = StartAggregatorAsync(entry); + + return subscription; + } + + /// + public IReadOnlyList GetCurrentAlarms(int siteId) + { + lock (_lock) + return _sites.TryGetValue(siteId, out var entry) ? entry.Current : Empty; + } + + /// + public bool IsLive(int siteId) + { + lock (_lock) + return _sites.TryGetValue(siteId, out var entry) && entry.HasPublished; + } + + // ── Subscriber teardown ───────────────────────────────────────────────────── + + private void Unsubscribe(Subscription subscription) + { + lock (_lock) + { + if (!_sites.TryGetValue(subscription.SiteId, out var entry)) + return; + + if (!entry.Subscribers.Remove(subscription)) + return; // already removed (idempotent Dispose) + + if (entry.Subscribers.Count == 0) + ScheduleLingerStop(entry); + } + } + + /// + /// Schedules a linger-delayed stop for a site whose last viewer just left. A + /// re-subscribe within the linger window cancels it (). + /// Must be called under . + /// + private void ScheduleLingerStop(SiteEntry entry) + { + var version = ++entry.LingerVersion; + entry.LingerTimer?.Dispose(); + entry.LingerTimer = new Timer( + _ => LingerStop(entry.SiteId, version), + state: null, + dueTime: _options.LiveAlarmCacheLinger, + period: Timeout.InfiniteTimeSpan); + } + + private void LingerStop(int siteId, int version) + { + IActorRef? actorToStop = null; + lock (_lock) + { + if (!_sites.TryGetValue(siteId, out var entry)) + return; + // A viewer may have returned (version bumped) or re-subscribed (subscribers > 0) + // after this timer was armed — only stop if this is still the latest, empty state. + if (entry.LingerVersion != version || entry.Subscribers.Count != 0) + return; + + actorToStop = entry.Actor; + entry.DisposeTimers(); + _sites.Remove(siteId); + } + + actorToStop?.Tell(new StopSiteAlarmAggregator()); + _logger.LogDebug("Stopped live alarm aggregator for site {SiteId} (last viewer left)", siteId); + } + + // ── Aggregator start ──────────────────────────────────────────────────────── + + private async Task StartAggregatorAsync(SiteEntry entry) + { + try + { + var system = _actorSystem; + if (system is null) + { + _logger.LogWarning( + "Live alarm cache asked to start site {SiteId} before the ActorSystem was set; " + + "the page will keep its poll fallback", entry.SiteId); + return; + } + + var resolved = await ResolveSiteAsync(entry.SiteId); + if (resolved is null) + return; // logged inside ResolveSiteAsync + + var (siteIdentifier, grpcA, grpcB) = resolved.Value; + var correlationId = $"site-alarm-{entry.SiteId}-{Guid.NewGuid():N}"; + + Func>> seedFn = + ct => FanOutSnapshotsAsync(entry.SiteId, ct); + Action> publish = + snapshot => OnPublish(entry, snapshot); + + lock (_lock) + { + // Bail if the entry was torn down (last viewer left + linger fired) while we + // resolved the site, or if a racing start already created the actor. + if (!_sites.TryGetValue(entry.SiteId, out var current) || !ReferenceEquals(current, entry)) + return; + if (entry.Actor is not null) + return; + + var props = Props.Create(() => new SiteAlarmAggregatorActor( + siteIdentifier, + correlationId, + seedFn, + publish, + _grpcFactory, + grpcA, + grpcB, + _options.LiveAlarmCacheReconcileInterval)); + + entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}"); + entry.StartRetryTimer?.Dispose(); + entry.StartRetryTimer = null; + _logger.LogInformation("Started live alarm aggregator for site {SiteId} ({SiteIdentifier})", + entry.SiteId, siteIdentifier); + } + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to start live alarm aggregator for site {SiteId}", entry.SiteId); + } + finally + { + lock (_lock) + { + entry.Starting = false; + + // A transient start failure (ActorSystem not yet set, a DB blip resolving + // the site, ActorOf throwing) would otherwise leave this viewer cohort + // polling forever — nothing re-attempts until the NEXT Subscribe. Arm a + // bounded retry so the feature self-heals once the condition clears, as + // long as viewers remain and the entry is still live. + var stillWanted = _sites.TryGetValue(entry.SiteId, out var current) + && ReferenceEquals(current, entry) + && entry.Actor is null + && entry.Subscribers.Count > 0; + if (stillWanted) + { + entry.Starting = true; // keep the guard closed so a Subscribe won't double-start + entry.StartRetryTimer?.Dispose(); + entry.StartRetryTimer = new Timer( + _ => { _ = StartAggregatorAsync(entry); }, + state: null, + dueTime: _options.LiveAlarmCacheReconcileInterval, + period: Timeout.InfiniteTimeSpan); + } + } + } + } + + /// + /// Resolves a site's identifier + both gRPC node endpoints. Returns null (and + /// logs) when the site is unknown or has no gRPC addresses configured, in which case + /// no aggregator is started and the page keeps polling. + /// + private async Task<(string SiteIdentifier, string GrpcA, string GrpcB)?> ResolveSiteAsync(int siteId) + { + using var scope = _serviceProvider.CreateScope(); + var siteRepo = scope.ServiceProvider.GetRequiredService(); + var site = await siteRepo.GetSiteByIdAsync(siteId); + if (site is null) + { + _logger.LogWarning("Live alarm cache: site {SiteId} not found; not starting an aggregator", siteId); + return null; + } + + if (string.IsNullOrWhiteSpace(site.GrpcNodeAAddress) || string.IsNullOrWhiteSpace(site.GrpcNodeBAddress)) + { + _logger.LogWarning( + "Live alarm cache: site {SiteId} ({SiteIdentifier}) has no gRPC node addresses; " + + "not starting an aggregator", siteId, site.SiteIdentifier); + return null; + } + + return (site.SiteIdentifier, site.GrpcNodeAAddress!, site.GrpcNodeBAddress!); + } + + // ── Snapshot fan-out (the seed / reconcile) ───────────────────────────────── + + /// + /// Fans out one debug-snapshot Ask per Enabled instance on the site (capped at + /// ) and returns every + /// snapshot's alarm rows — including + /// placeholders. Best-effort: an instance that faults, times out, or reports + /// contributes no rows rather than + /// failing the whole seed (mirrors AlarmSummaryService). Re-enumerates instances + /// each call so a reconcile picks up enable/disable/deploy/delete drift. + /// + private async Task> FanOutSnapshotsAsync(int siteId, CancellationToken ct) + { + string siteIdentifier; + List enabledInstanceNames; + + using (var scope = _serviceProvider.CreateScope()) + { + var siteRepo = scope.ServiceProvider.GetRequiredService(); + var site = await siteRepo.GetSiteByIdAsync(siteId, ct); + if (site is null) + return Empty; + siteIdentifier = site.SiteIdentifier; + + var instanceRepo = scope.ServiceProvider.GetRequiredService(); + var instances = await instanceRepo.GetInstancesBySiteIdAsync(siteId, ct); + enabledInstanceNames = instances + .Where(i => i.State == InstanceState.Enabled) + .Select(i => i.UniqueName) + .ToList(); + } + + if (enabledInstanceNames.Count == 0) + return Empty; + + var rows = new ConcurrentBag(); + using var gate = new SemaphoreSlim( + Math.Max(1, _options.LiveAlarmCacheSeedConcurrency), + Math.Max(1, _options.LiveAlarmCacheSeedConcurrency)); + + var fetches = enabledInstanceNames.Select(name => + FetchInstanceSnapshotAsync(siteIdentifier, name, gate, rows, ct)); + await Task.WhenAll(fetches); + + return rows.ToList(); + } + + private async Task FetchInstanceSnapshotAsync( + string siteIdentifier, + string instanceUniqueName, + SemaphoreSlim gate, + ConcurrentBag rows, + CancellationToken ct) + { + await gate.WaitAsync(ct); + try + { + var request = new DebugSnapshotRequest(instanceUniqueName, Guid.NewGuid().ToString("N")); + var snapshot = await _communicationService.RequestDebugSnapshotAsync(siteIdentifier, request, ct); + if (snapshot.InstanceNotFound) + return; + + foreach (var alarm in snapshot.AlarmStates) + rows.Add(alarm); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; // seed cancelled (actor stopping) — propagate + } + catch + { + // Degrade this one instance to "no rows" rather than failing the whole seed. + } + finally + { + gate.Release(); + } + } + + // ── Publish (actor thread → viewers) ──────────────────────────────────────── + + private void OnPublish(SiteEntry entry, IReadOnlyList snapshot) + { + Subscription[] subscribers; + lock (_lock) + { + // Store the fresh immutable snapshot (readers get it lock-free-ish via GetCurrentAlarms). + entry.Current = snapshot; + entry.HasPublished = true; + subscribers = entry.Subscribers.ToArray(); + } + + // Invoke viewer callbacks OUTSIDE the lock — a Blazor InvokeAsync(StateHasChanged) + // must never re-enter our lock, and one faulting viewer must not stop the others. + foreach (var subscriber in subscribers) + { + try + { + subscriber.OnChanged(); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "A live alarm viewer callback threw for site {SiteId}; continuing", entry.SiteId); + } + } + } + + // ── Nested types ──────────────────────────────────────────────────────────── + + /// Per-site bookkeeping. All mutable fields are guarded by the service's _lock. + private sealed class SiteEntry + { + public SiteEntry(int siteId) => SiteId = siteId; + + public int SiteId { get; } + public List Subscribers { get; } = new(); + public IActorRef? Actor { get; set; } + public bool Starting { get; set; } + + /// Current published snapshot the page renders. Swapped as a whole immutable list. + public IReadOnlyList Current { get; set; } = Empty; + + /// True once the aggregator has seeded and published at least once. + public bool HasPublished { get; set; } + + public Timer? LingerTimer { get; set; } + public int LingerVersion { get; set; } + + /// Bounded retry timer armed when a start attempt failed to create the actor. + public Timer? StartRetryTimer { get; set; } + + /// Cancels any pending linger stop (a viewer arrived). + public void CancelLinger() + { + LingerVersion++; + LingerTimer?.Dispose(); + LingerTimer = null; + } + + /// Disposes both timers — called when the entry is torn down. + public void DisposeTimers() + { + LingerTimer?.Dispose(); + LingerTimer = null; + StartRetryTimer?.Dispose(); + StartRetryTimer = null; + } + } + + /// The disposable handed to a viewer; idempotent unregisters it. + private sealed class Subscription : IDisposable + { + private readonly SiteAlarmLiveCacheService _owner; + private int _disposed; + + public Subscription(SiteAlarmLiveCacheService owner, int siteId, Action onChanged) + { + _owner = owner; + SiteId = siteId; + OnChanged = onChanged; + } + + public int SiteId { get; } + public Action OnChanged { get; } + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) == 0) + _owner.Unsubscribe(this); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index 99141f60..7f060921 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -424,6 +424,11 @@ akka {{ var debugStreamService = _serviceProvider.GetService(); debugStreamService?.SetActorSystem(_actorSystem!); + // Wire up the aggregated live alarm cache with the ActorSystem so it can host + // per-site aggregator actors on the active central node (plan #10, Task 4). + var siteAlarmLiveCache = _serviceProvider.GetService(); + siteAlarmLiveCache?.SetActorSystem(_actorSystem!); + // Management Service — accessible via ClusterClient var mgmtLogger = _serviceProvider.GetRequiredService() diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs new file mode 100644 index 00000000..32846846 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -0,0 +1,414 @@ +using System.Collections.Concurrent; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; + +/// +/// Tests for (plan #10, Task 4) — the transient, +/// in-memory, per-site live alarm aggregator. Covers the tricky bits: seed-then-stream +/// ordering (a delta arriving before the seed completes is neither lost nor double-applied), +/// dedup by AlarmKey, placeholder seeded-but-not-on-stream coherence, NodeA↔NodeB reconnect +/// re-seed, and periodic reconcile authoritative-replace (drift correction). +/// +public class SiteAlarmAggregatorActorTests : TestKit +{ + private const string SiteId = "site-alpha"; + private const string GrpcNodeA = "http://localhost:5100"; + private const string GrpcNodeB = "http://localhost:5200"; + private const string Instance = "Site1.Pump01"; + + public SiteAlarmAggregatorActorTests() : base(@"akka.loglevel = WARNING") + { + SiteAlarmAggregatorActor.ReconnectDelay = TimeSpan.FromMilliseconds(50); + SiteAlarmAggregatorActor.StabilityWindow = TimeSpan.FromSeconds(30); + } + + // ── Test doubles ──────────────────────────────────────────────────────────── + + /// + /// Controllable seed fan-out: each invocation returns a fresh Task the test completes + /// on demand (so seed/reconcile timing relative to buffered deltas is deterministic). + /// + private sealed class SeedStub + { + private readonly ConcurrentQueue>> _pending = new(); + private int _callCount; + + public int CallCount => Volatile.Read(ref _callCount); + + public Task> Seed(CancellationToken ct) + { + Interlocked.Increment(ref _callCount); + var tcs = new TaskCompletionSource>( + TaskCreationOptions.RunContinuationsAsynchronously); + ct.Register(() => tcs.TrySetCanceled()); + _pending.Enqueue(tcs); + return tcs.Task; + } + + /// Completes the oldest not-yet-completed seed call with the given rows. + public void CompleteNext(params AlarmStateChanged[] rows) + { + SpinWait.SpinUntil(() => _pending.TryPeek(out _), TimeSpan.FromSeconds(3)); + if (_pending.TryDequeue(out var tcs)) + tcs.TrySetResult(rows); + else + throw new InvalidOperationException("No pending seed to complete."); + } + + /// Faults the oldest not-yet-completed seed call (simulates a whole-fan-out failure). + public void FaultNext() + { + SpinWait.SpinUntil(() => _pending.TryPeek(out _), TimeSpan.FromSeconds(3)); + if (_pending.TryDequeue(out var tcs)) + tcs.TrySetException(new InvalidOperationException("seed boom")); + else + throw new InvalidOperationException("No pending seed to fault."); + } + } + + private sealed class PublishSink + { + private readonly object _lock = new(); + public List> Snapshots { get; } = new(); + + public void Publish(IReadOnlyList snapshot) + { + lock (_lock) { Snapshots.Add(snapshot); } + } + + public IReadOnlyList? Latest + { + get { lock (_lock) { return Snapshots.Count == 0 ? null : Snapshots[^1]; } } + } + + public int Count { get { lock (_lock) { return Snapshots.Count; } } } + } + + private sealed record SiteSub( + string CorrelationId, Action OnAlarm, Action OnError, CancellationToken Ct); + + private sealed class MockSiteAlarmStreamClient : SiteStreamGrpcClient + { + private readonly object _lock = new(); + private readonly List _subs = new(); + private readonly List _unsubscribed = new(); + + public List Subs { get { lock (_lock) { return _subs.ToList(); } } } + public List Unsubscribed { get { lock (_lock) { return _unsubscribed.ToList(); } } } + + public MockSiteAlarmStreamClient() : base() { } + + public override Task SubscribeSiteAsync( + string correlationId, Action onAlarmEvent, Action onError, CancellationToken ct) + { + lock (_lock) { _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, ct)); } + var tcs = new TaskCompletionSource(); + ct.Register(() => tcs.TrySetResult()); + return tcs.Task; // never completes until cancelled (simulates a live stream) + } + + public override void Unsubscribe(string correlationId) + { + lock (_lock) { _unsubscribed.Add(correlationId); } + } + } + + private sealed class MockSiteAlarmStreamClientFactory : SiteStreamGrpcClientFactory + { + private readonly ConcurrentDictionary _byEndpoint = new(); + + public MockSiteAlarmStreamClientFactory() + : base(Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance) { } + + public MockSiteAlarmStreamClient ClientFor(string endpoint) => + _byEndpoint.GetOrAdd(endpoint, _ => new MockSiteAlarmStreamClient()); + + public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint) + => ClientFor(grpcEndpoint); + + public override SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint) + => _byEndpoint.TryGetValue(grpcEndpoint, out var c) ? c : null; + } + + private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor( + TimeSpan? reconcileInterval = null) + { + var seed = new SeedStub(); + var sink = new PublishSink(); + var factory = new MockSiteAlarmStreamClientFactory(); + + var props = Props.Create(() => new SiteAlarmAggregatorActor( + SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB, + reconcileInterval ?? TimeSpan.FromMinutes(10))); + + var actor = Sys.ActorOf(props); + return (actor, seed, sink, factory); + } + + private static AlarmStateChanged Alarm( + string alarmName, string sourceRef, int priority, DateTimeOffset ts, + bool placeholder = false, string instance = Instance) => + new(instance, alarmName, AlarmState.Active, priority, ts) + { + SourceReference = sourceRef, + IsConfiguredPlaceholder = placeholder, + Kind = placeholder ? AlarmKind.NativeOpcUa : AlarmKind.Computed + }; + + private static AlarmStateChanged? Find(IReadOnlyList? rows, string alarmName, string sourceRef) => + rows?.FirstOrDefault(r => r.AlarmName == alarmName && r.SourceReference == sourceRef); + + // ── Tests ─────────────────────────────────────────────────────────────────── + + [Fact] + public void Delta_Arriving_Before_Seed_Completes_Is_Buffered_And_Applied_Exactly_Once() + { + var (actor, seed, sink, _) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + + // Live delta arrives DURING the seed window (new key, not in the seed). + var t1 = DateTimeOffset.UtcNow; + actor.Tell(Alarm("Overheat", "", 900, t1)); + + // Still buffering — nothing published yet. + Assert.Equal(0, sink.Count); + + // Seed completes with a DIFFERENT alarm. + seed.CompleteNext(Alarm("PumpFault", "", 500, t1.AddSeconds(-1))); + + AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3)); + var latest = sink.Latest!; + Assert.NotNull(Find(latest, "PumpFault", "")); // seeded + var overheat = Find(latest, "Overheat", ""); // buffered gap-window delta survived + Assert.NotNull(overheat); + Assert.Single(latest, r => r.AlarmName == "Overheat"); // exactly once + } + + [Fact] + public void Buffered_Delta_Older_Than_Seed_Entry_Is_Dropped() + { + var (actor, seed, sink, _) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + + var tSeed = DateTimeOffset.UtcNow; + // Buffered delta for the SAME key but OLDER than the seed entry. + actor.Tell(Alarm("PumpFault", "", 100, tSeed.AddSeconds(-1))); + seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed)); + + AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3)); + // Give a beat to ensure no late (dropped) delta sneaks in. + Thread.Sleep(150); + var latest = sink.Latest!; + var pumpFault = Find(latest, "PumpFault", ""); + Assert.NotNull(pumpFault); + Assert.Equal(500, pumpFault!.Priority); // seed value kept; older buffered delta dropped + } + + [Fact] + public void Buffered_Delta_Newer_Than_Seed_Entry_Is_Applied() + { + var (actor, seed, sink, _) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + + var tSeed = DateTimeOffset.UtcNow; + actor.Tell(Alarm("PumpFault", "", 900, tSeed.AddSeconds(1))); // newer than seed + seed.CompleteNext(Alarm("PumpFault", "", 500, tSeed)); + + AwaitCondition(() => sink.Latest is { } l && Find(l, "PumpFault", "")?.Priority == 900, + TimeSpan.FromSeconds(3)); + } + + [Fact] + public void Dedup_By_AlarmKey_Distinguishes_SourceReference() + { + var (actor, seed, sink, _) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); // empty seed + AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3)); + + var t = DateTimeOffset.UtcNow; + // Same instance + alarm name, different source reference → two distinct rows. + actor.Tell(Alarm("LevelAlarm", "Tank01.Level.HiHi", 700, t)); + actor.Tell(Alarm("LevelAlarm", "Tank01.Level.LoLo", 300, t)); + + AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3)); + var latest = sink.Latest!; + Assert.NotNull(Find(latest, "LevelAlarm", "Tank01.Level.HiHi")); + Assert.NotNull(Find(latest, "LevelAlarm", "Tank01.Level.LoLo")); + } + + [Fact] + public void Same_Key_Live_Update_Replaces_In_Place() + { + var (actor, seed, sink, _) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3)); + + var t = DateTimeOffset.UtcNow; + actor.Tell(Alarm("PumpFault", "", 500, t)); + AwaitCondition(() => Find(sink.Latest, "PumpFault", "")?.Priority == 500, TimeSpan.FromSeconds(3)); + + // Newer update for the same key replaces (not duplicates). + actor.Tell(Alarm("PumpFault", "", 800, t.AddSeconds(1))); + AwaitCondition(() => Find(sink.Latest, "PumpFault", "")?.Priority == 800, TimeSpan.FromSeconds(3)); + Assert.Single(sink.Latest!); // still exactly one row for the key + } + + [Fact] + public void Seeded_Native_Alarm_And_Its_Live_Delta_Collapse_Onto_One_Key() + { + // Identity parity across the two paths (snapshot seed vs live stream): a native + // alarm seeded with a non-empty SourceReference and a later live delta for the + // SAME (instance, alarmName, sourceRef) must collapse to ONE in-place-updated row, + // not a ghost duplicate. Both paths key on SourceReference, so they must agree. + var (actor, seed, sink, _) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + + var t = DateTimeOffset.UtcNow; + seed.CompleteNext(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 300, t)); + AwaitCondition(() => Find(sink.Latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi") is not null, + TimeSpan.FromSeconds(3)); + + // Live delta for the exact same identity, newer → updates in place. + actor.Tell(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 800, t.AddSeconds(1))); + + AwaitCondition( + () => Find(sink.Latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi")?.Priority == 800, + TimeSpan.FromSeconds(3)); + Assert.Single(sink.Latest!); // exactly one row — no seed/live ghost duplicate + } + + [Fact] + public void Placeholder_Seeded_Coexists_With_Live_Real_Alarm_And_Is_Not_Wiped() + { + var (actor, seed, sink, _) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + + var t = DateTimeOffset.UtcNow; + // Seed carries a configured-placeholder row (binding with no active conditions). + var placeholder = Alarm("Motor1.MotorAlarms", "", 0, t, placeholder: true); + seed.CompleteNext(placeholder); + AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3)); + + // A live REAL condition for the same instance arrives (distinct AlarmKey). + actor.Tell(Alarm("Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi", 800, t.AddSeconds(1))); + + AwaitCondition(() => sink.Latest is { } l && l.Count == 2, TimeSpan.FromSeconds(3)); + var latest = sink.Latest!; + // Placeholder still present (not wiped by the real alarm under a different key). + var ph = Find(latest, "Motor1.MotorAlarms", ""); + Assert.NotNull(ph); + Assert.True(ph!.IsConfiguredPlaceholder); + Assert.NotNull(Find(latest, "Motor1.MotorAlarms", "Motor1.MotorAlarms.HiHi")); + } + + [Fact] + public void GrpcError_Flips_Node_Reconnects_And_ReSeeds() + { + var (_, seed, sink, factory) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); // initial seed done + // Wait for the seed to actually be APPLIED (published) so the re-seed on error is + // not skipped as "fan-out already in flight". + AwaitCondition(() => sink.Count >= 1, TimeSpan.FromSeconds(3)); + + // Fail the NodeA stream. + factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("NodeA down")); + + // Reconnect must reach NodeB, AND a re-seed fan-out must have been triggered. + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); + AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5)); + // The failed NodeA stream was unsubscribed (relay released, not left zombie). + Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).Unsubscribed); + } + + [Fact] + public void Periodic_Reconcile_Authoritatively_Replaces_Stale_Rows() + { + var (_, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(250)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + + var t = DateTimeOffset.UtcNow; + seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // seed has a row + AwaitCondition(() => Find(sink.Latest, "PumpFault", "") is not null, TimeSpan.FromSeconds(3)); + + // Reconcile tick fires → seed #2 (e.g. instance was disabled) returns EMPTY. + AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); // empty + + // The stale PumpFault row must disappear (authoritative replace, not merge). + AwaitCondition(() => sink.Latest is { } l && l.Count == 0, TimeSpan.FromSeconds(3)); + } + + [Fact] + public void Failed_Initial_Seed_Does_Not_Publish_But_A_Later_Reconcile_Recovers() + { + var (_, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(250)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + + // The initial seed fan-out throws as a whole. + seed.FaultNext(); + + // A failed INITIAL seed must not publish (page keeps its poll fallback — not live yet). + Thread.Sleep(200); + Assert.Equal(0, sink.Count); + + // The periodic reconcile is the backstop: it re-runs the fan-out (flag was cleared + // on failure), completes, and the actor publishes → now live. + AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); + seed.CompleteNext(Alarm("PumpFault", "", 500, DateTimeOffset.UtcNow)); + AwaitCondition(() => Find(sink.Latest, "PumpFault", "") is not null, TimeSpan.FromSeconds(3)); + } + + [Fact] + public void Stream_GivenUp_After_MaxRetries_Reopens_On_Reconcile_Tick() + { + // After the retry budget is exhausted the live stream is left down, but the + // aggregator is NOT stopped — the next reconcile tick self-heals it (resets the + // budget + reopens) so a sustained site outage never permanently kills the feed. + var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(300)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + + // Drive initial + 3 retries (each flips node) to exhaust the budget. + factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1")); + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); + factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2")); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5)); + factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3")); + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5)); + + int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count; + var before = TotalSubs(); // 4 + + // Fourth error exceeds MaxRetries → stream given up (no immediate reopen). + factory.ClientFor(GrpcNodeB).Subs[1].OnError(new Exception("4")); + + // The reconcile tick reopens the stream (self-heal). + AwaitCondition(() => TotalSubs() > before, TimeSpan.FromSeconds(3)); + } + + [Fact] + public void Stop_Message_TearsDown_Grpc_And_Stops_Actor() + { + var (actor, seed, _, factory) = CreateActor(); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + + Watch(actor); + actor.Tell(new StopSiteAlarmAggregator()); + + ExpectTerminated(actor, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Unsubscribed.Contains("corr-1"), + TimeSpan.FromSeconds(3)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs new file mode 100644 index 00000000..40d56b68 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs @@ -0,0 +1,227 @@ +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +/// +/// Tests for (plan #10, Task 4) — the DI singleton +/// that reference-counts Alarm Summary viewers over one shared per-site aggregator. Covers +/// shared start (one aggregator for many viewers), last-out linger stop, and re-subscribe +/// cancelling a pending stop. Uses a real (in-memory) ActorSystem and a mock gRPC factory +/// whose site-wide stream never faults, so no reconnect noise. +/// +public class SiteAlarmLiveCacheServiceTests : TestKit +{ + private const int SiteId = 7; + + // A mock site-wide alarm stream client whose subscription hangs until cancelled. + private sealed class HangingClient : SiteStreamGrpcClient + { + public HangingClient() : base() { } + public override Task SubscribeSiteAsync( + string correlationId, Action onAlarmEvent, + Action onError, CancellationToken ct) + { + var tcs = new TaskCompletionSource(); + ct.Register(() => tcs.TrySetResult()); + return tcs.Task; + } + public override void Unsubscribe(string correlationId) { } + } + + private sealed class CountingFactory : SiteStreamGrpcClientFactory + { + private readonly HangingClient _client = new(); + public int GetOrCreateCount; + + public CountingFactory() : base(NullLoggerFactory.Instance) { } + + public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint) + { + Interlocked.Increment(ref GetOrCreateCount); + return _client; + } + + public override SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint) => _client; + } + + private SiteAlarmLiveCacheService CreateService(TimeSpan linger, out CountingFactory factory) + { + // Site with gRPC addresses, and NO enabled instances → the seed fan-out returns + // empty immediately (so IsLive flips true fast without any snapshot Asks). + var site = new Site("Alpha", "site-alpha") + { + Id = SiteId, + GrpcNodeAAddress = "http://localhost:5100", + GrpcNodeBAddress = "http://localhost:5200" + }; + + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(SiteId, Arg.Any()).Returns(site); + + var instanceRepo = Substitute.For(); + instanceRepo.GetInstancesBySiteIdAsync(SiteId, Arg.Any()) + .Returns(new List()); + + var services = new ServiceCollection(); + services.AddScoped(_ => siteRepo); + services.AddScoped(_ => instanceRepo); + var provider = services.BuildServiceProvider(); + + var options = Options.Create(new CommunicationOptions + { + LiveAlarmCacheLinger = linger, + LiveAlarmCacheReconcileInterval = TimeSpan.FromMinutes(10) + }); + + var comm = new CommunicationService(Options.Create(new CommunicationOptions()), + NullLogger.Instance); + + factory = new CountingFactory(); + var service = new SiteAlarmLiveCacheService( + provider, comm, factory, options, NullLogger.Instance); + service.SetActorSystem(Sys); + return service; + } + + [Fact] + public void First_Subscriber_Starts_One_Aggregator_Shared_By_Multiple_Viewers() + { + var service = CreateService(TimeSpan.FromMilliseconds(200), out var factory); + + var changes1 = 0; + var changes2 = 0; + using var sub1 = service.Subscribe(SiteId, () => Interlocked.Increment(ref changes1)); + using var sub2 = service.Subscribe(SiteId, () => Interlocked.Increment(ref changes2)); + + // The (empty) seed completes → aggregator publishes → live. + AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5)); + + // Exactly ONE gRPC stream opened for the site regardless of viewer count. + Assert.Equal(1, factory.GetOrCreateCount); + // Empty seed → empty current snapshot. + Assert.Empty(service.GetCurrentAlarms(SiteId)); + // Both viewers were notified of the initial publish. + AwaitCondition(() => changes1 >= 1 && changes2 >= 1, TimeSpan.FromSeconds(3)); + } + + [Fact] + public void Last_Viewer_Leaving_Stops_Aggregator_After_Linger() + { + var service = CreateService(TimeSpan.FromMilliseconds(200), out _); + + var sub1 = service.Subscribe(SiteId, () => { }); + var sub2 = service.Subscribe(SiteId, () => { }); + AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5)); + + sub1.Dispose(); + // One viewer remains — still live, not stopped. + Thread.Sleep(300); + Assert.True(service.IsLive(SiteId)); + + sub2.Dispose(); + // Last viewer left → after the linger the aggregator is torn down and the entry cleared. + AwaitCondition(() => !service.IsLive(SiteId), TimeSpan.FromSeconds(3)); + Assert.Empty(service.GetCurrentAlarms(SiteId)); + } + + [Fact] + public void Resubscribe_Within_Linger_Cancels_The_Pending_Stop() + { + var service = CreateService(TimeSpan.FromMilliseconds(400), out var factory); + + var sub1 = service.Subscribe(SiteId, () => { }); + AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5)); + + sub1.Dispose(); // schedules linger stop + var sub2 = service.Subscribe(SiteId, () => { }); // returns within the linger window → cancels it + + // Wait comfortably past the original linger; the aggregator must still be alive. + Thread.Sleep(700); + Assert.True(service.IsLive(SiteId)); + // No second aggregator/stream was opened — the same one was kept warm. + Assert.Equal(1, factory.GetOrCreateCount); + + sub2.Dispose(); + } + + [Fact] + public void Dispose_Is_Idempotent() + { + var service = CreateService(TimeSpan.FromMilliseconds(200), out _); + var sub = service.Subscribe(SiteId, () => { }); + AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5)); + + sub.Dispose(); + sub.Dispose(); // must not throw or double-decrement + + AwaitCondition(() => !service.IsLive(SiteId), TimeSpan.FromSeconds(3)); + } + + [Fact] + public void Transient_Start_Failure_Self_Heals_On_Retry_For_A_Stable_Viewer_Cohort() + { + // A transient failure resolving the site at first-subscribe must NOT leave the + // viewer cohort polling forever — a bounded retry re-attempts and self-heals. + var site = new Site("Alpha", "site-alpha") + { + Id = SiteId, + GrpcNodeAAddress = "http://localhost:5100", + GrpcNodeBAddress = "http://localhost:5200" + }; + + var siteRepo = Substitute.For(); + // First resolve returns null (transient blip); every resolve thereafter succeeds. + siteRepo.GetSiteByIdAsync(SiteId, Arg.Any()).Returns((Site?)null, site); + + var instanceRepo = Substitute.For(); + instanceRepo.GetInstancesBySiteIdAsync(SiteId, Arg.Any()) + .Returns(new List()); + + var services = new ServiceCollection(); + services.AddScoped(_ => siteRepo); + services.AddScoped(_ => instanceRepo); + var provider = services.BuildServiceProvider(); + + var options = Options.Create(new CommunicationOptions + { + LiveAlarmCacheLinger = TimeSpan.FromSeconds(30), + // Retry cadence reuses the reconcile interval — keep it short for the test. + LiveAlarmCacheReconcileInterval = TimeSpan.FromMilliseconds(300) + }); + var comm = new CommunicationService(Options.Create(new CommunicationOptions()), + NullLogger.Instance); + var factory = new CountingFactory(); + + var service = new SiteAlarmLiveCacheService( + provider, comm, factory, options, NullLogger.Instance); + service.SetActorSystem(Sys); + + using var sub = service.Subscribe(SiteId, () => { }); + + // First start failed (site not found); a stable single viewer stays subscribed. + // The retry eventually resolves the site and the aggregator goes live — without + // any further Subscribe call. + AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5)); + } + + [Fact] + public void Unknown_Site_Registers_Viewer_But_Never_Goes_Live() + { + var service = CreateService(TimeSpan.FromMilliseconds(200), out _); + + using var sub = service.Subscribe(999, () => { }); // no such site in the repo + Thread.Sleep(300); + Assert.False(service.IsLive(999)); + Assert.Empty(service.GetCurrentAlarms(999)); + } +}