feat(sitestream): central transient per-site live alarm cache w/ seed-then-stream + failover (plan #10 T4)
Adds the active-central-node, in-memory, reference-counted per-site live alarm cache backing the operator Alarm Summary page. No persisted central alarm store ([PERM]) — no EF entity/table/migration/DbSet. - ISiteAlarmLiveCache (new): singleton seam — Subscribe(siteId, onChanged) -> IDisposable (ref-counted, linger stop on last-out), GetCurrentAlarms, IsLive. - SiteAlarmAggregatorActor (new): one per site. Seed-then-stream ordering copied from DebugStreamBridgeActor — open the site-wide alarm-only gRPC stream first, buffer live deltas while the snapshot fan-out runs, flush with per-key dedup (AlarmKey = InstanceUniqueName|AlarmName|SourceReference), then live pass-through into an in-memory dict. Placeholder rows seeded from the snapshot and never expected on the stream; a real-alarm delta (distinct key) never wipes them. NodeA<->NodeB reconnect (retry budget + stability window), reconnect re-seeds, periodic reconcile (authoritative clear-and-rebuild) corrects instance-set drift, and a budget-exhausted stream self-heals on the next reconcile tick. - SiteAlarmLiveCacheService (new): DI singleton facade — viewer reference-counting, linger-delayed last-out stop (version + TryRemove(ref) race guards), the snapshot fan-out seed (RequestDebugSnapshotAsync per Enabled instance, capped), bounded start-retry self-heal on transient start failure, and the immutable published-snapshot store the page reads. Cache mutated only on the actor thread; viewer callbacks invoked outside the lock. - CommunicationOptions: LiveAlarmCacheLinger (30s), LiveAlarmCacheReconcileInterval (60s), LiveAlarmCacheSeedConcurrency (8), LiveAlarmCacheMaxSubscribersPerSite (200). Task 6 formalizes eager validation + telemetry. - DI registration + AkkaHostedService SetActorSystem wiring on the active central node. - Tests: 14 actor (seed/stream ordering, dedup, native-alarm parity, placeholder coherence, reconnect re-seed, reconcile replace, self-heal, stop) + 6 service (shared start, linger stop, resubscribe cancels stop, idempotent dispose, unknown site, transient-start self-heal). Code-reviewer pass: no persistence, no Critical; both Important findings addressed. Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
This commit is contained in:
Joseph Doherty
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
/// <c>Dictionary<AlarmKey, AlarmStateChanged></c> of the whole site's current
|
||||
/// alarm state — NO persistence (locked <c>[PERM]</c>: no central alarm store). Created
|
||||
/// and torn down by <see cref="SiteAlarmLiveCacheService"/> under viewer reference-count.
|
||||
/// <para>
|
||||
/// <b>Seed-then-stream ordering</b> (copied from <see cref="DebugStreamBridgeActor"/>):
|
||||
/// the site-wide, alarm-only gRPC stream (<c>SubscribeSite</c>) is opened FIRST in
|
||||
/// <see cref="PreStart"/> 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
|
||||
/// <em>buffered in arrival order</em>. 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Placeholder reconciliation:</b> the snapshot fan-out carries
|
||||
/// <see cref="AlarmStateChanged.IsConfiguredPlaceholder"/> 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 <c>AlarmKey</c> so it can never wipe a placeholder row and
|
||||
/// vice-versa. Placeholder-vs-real coherence is refreshed by the periodic reconcile.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Failover + drift:</b> a gRPC error flips NodeA↔NodeB with the same retry budget +
|
||||
/// stability window as <see cref="DebugStreamBridgeActor"/>, and each reconnect triggers
|
||||
/// a RE-SEED (never silently serve stale). A periodic reconcile snapshot
|
||||
/// (<see cref="_reconcileInterval"/>, default 60s) corrects instance-set drift and any
|
||||
/// missed delta.
|
||||
/// </para>
|
||||
/// All state is mutated only on the actor thread: gRPC callbacks and fan-out results are
|
||||
/// marshalled back via <c>Self.Tell</c>, 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.
|
||||
/// </summary>
|
||||
public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
|
||||
{
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
private readonly string _siteIdentifier;
|
||||
private readonly string _correlationId;
|
||||
private readonly Func<CancellationToken, Task<IReadOnlyList<AlarmStateChanged>>> _seedFn;
|
||||
private readonly Action<IReadOnlyList<AlarmStateChanged>> _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";
|
||||
|
||||
/// <summary>Delay between gRPC reconnection attempts. Settable for tests.</summary>
|
||||
internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
|
||||
|
||||
/// <summary>
|
||||
/// How long a freshly-opened gRPC stream must stay up before its retry budget is
|
||||
/// considered recovered (mirrors <see cref="DebugStreamBridgeActor.StabilityWindow"/>).
|
||||
/// Settable for tests.
|
||||
/// </summary>
|
||||
internal static TimeSpan StabilityWindow { get; set; } = TimeSpan.FromSeconds(60);
|
||||
|
||||
private int _retryCount;
|
||||
private bool _useNodeA = true;
|
||||
private bool _stopped;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private bool _streamDown;
|
||||
private CancellationTokenSource? _grpcCts;
|
||||
private CancellationTokenSource? _lifetimeCts;
|
||||
|
||||
/// <summary>Current whole-site alarm state, keyed by <see cref="AlarmKey"/>. Actor-thread only.</summary>
|
||||
private readonly Dictionary<string, AlarmStateChanged> _cache = new();
|
||||
|
||||
/// <summary>True once the first seed has completed and been published. Actor-thread only.</summary>
|
||||
private bool _seeded;
|
||||
|
||||
/// <summary>True while a seed/reconcile snapshot fan-out is in flight (deltas buffer). Actor-thread only.</summary>
|
||||
private bool _fanoutInFlight;
|
||||
|
||||
/// <summary>Ordered buffer of live deltas that arrived while a fan-out was in flight. Actor-thread only.</summary>
|
||||
private readonly List<AlarmStateChanged> _buffer = new();
|
||||
|
||||
private const int BufferWarnThreshold = 10_000;
|
||||
private bool _bufferWarned;
|
||||
|
||||
/// <inheritdoc />
|
||||
public ITimerScheduler Timers { get; set; } = null!;
|
||||
|
||||
/// <summary>
|
||||
/// Creates a per-site alarm aggregator.
|
||||
/// </summary>
|
||||
/// <param name="siteIdentifier">Site identifier (for logging / gRPC client keying).</param>
|
||||
/// <param name="correlationId">Correlation id for the site-wide gRPC subscription.</param>
|
||||
/// <param name="seedFn">
|
||||
/// 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.
|
||||
/// </param>
|
||||
/// <param name="publish">
|
||||
/// Publishes a fresh immutable snapshot of the cache to the owning service, which
|
||||
/// stores it and raises viewer <c>onChanged</c> callbacks. Invoked on the actor thread.
|
||||
/// </param>
|
||||
/// <param name="grpcFactory">Factory caching one gRPC client per (site, endpoint).</param>
|
||||
/// <param name="grpcNodeAAddress">gRPC address of the site's node A.</param>
|
||||
/// <param name="grpcNodeBAddress">gRPC address of the site's node B.</param>
|
||||
/// <param name="reconcileInterval">Periodic reconcile snapshot cadence.</param>
|
||||
public SiteAlarmAggregatorActor(
|
||||
string siteIdentifier,
|
||||
string correlationId,
|
||||
Func<CancellationToken, Task<IReadOnlyList<AlarmStateChanged>>> seedFn,
|
||||
Action<IReadOnlyList<AlarmStateChanged>> 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<AlarmStateChanged>(HandleLiveDelta);
|
||||
|
||||
// A seed/reconcile fan-out completed.
|
||||
Receive<SeedCompleted>(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<SeedFailed>(OnSeedFailed);
|
||||
|
||||
// Periodic reconcile tick (and the re-seed kicked after a reconnect).
|
||||
Receive<RunReconcile>(_ => OnReconcileTick());
|
||||
|
||||
// Stream stayed up for StabilityWindow — recover the retry budget.
|
||||
Receive<GrpcAlarmStreamStable>(_ =>
|
||||
{
|
||||
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<GrpcAlarmStreamError>(msg =>
|
||||
{
|
||||
_log.Warning("Site-alarm gRPC stream error for {0}: {1}", _siteIdentifier, msg.Exception.Message);
|
||||
HandleGrpcError();
|
||||
});
|
||||
|
||||
Receive<ReconnectAlarmStream>(_ => OpenGrpcStream());
|
||||
|
||||
// Owning service asks us to stop (last viewer left + linger elapsed).
|
||||
Receive<StopSiteAlarmAggregator>(_ =>
|
||||
{
|
||||
_log.Info("Stopping site-alarm aggregator for {0}", _siteIdentifier);
|
||||
CleanupGrpc();
|
||||
_stopped = true;
|
||||
Context.Stop(Self);
|
||||
});
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void PostStop()
|
||||
{
|
||||
_grpcCts?.Cancel();
|
||||
_grpcCts?.Dispose();
|
||||
_grpcCts = null;
|
||||
_lifetimeCts?.Cancel();
|
||||
_lifetimeCts?.Dispose();
|
||||
_lifetimeCts = null;
|
||||
base.PostStop();
|
||||
}
|
||||
|
||||
// ── Reconcile tick ──────────────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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 ────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Kicks a snapshot fan-out as a background task, marshalling the result back via
|
||||
/// <c>Self.Tell</c>. While in flight, live deltas buffer. A reconcile that arrives
|
||||
/// while a fan-out is already running is skipped (no stacking).
|
||||
/// </summary>
|
||||
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();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Flushes the pre-fan-out buffer in arrival order. When <paramref name="dedupAgainstSeed"/>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Applies one alarm delta to the cache keyed by <see cref="AlarmKey"/>. Returns
|
||||
/// <c>true</c> 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.
|
||||
/// </summary>
|
||||
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) ──
|
||||
|
||||
/// <summary>
|
||||
/// NUL delimiter so distinct identities never collide on a shared boundary. Cannot
|
||||
/// appear in an instance/alarm name. Mirrors <see cref="DebugStreamBridgeActor"/>.
|
||||
/// </summary>
|
||||
private const char KeyDelimiter = '\u0000';
|
||||
|
||||
/// <summary>
|
||||
/// Per-alarm dedup identity = (InstanceUniqueName, AlarmName, SourceReference) —
|
||||
/// identical to <c>DebugStreamBridgeActor.AlarmKey</c> 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.
|
||||
/// </summary>
|
||||
private static string AlarmKey(AlarmStateChanged a) =>
|
||||
string.Concat(
|
||||
a.InstanceUniqueName ?? string.Empty, KeyDelimiter,
|
||||
a.AlarmName ?? string.Empty, KeyDelimiter,
|
||||
a.SourceReference ?? string.Empty);
|
||||
}
|
||||
|
||||
/// <summary>Message asking a <see cref="SiteAlarmAggregatorActor"/> to stop (last viewer left).</summary>
|
||||
public sealed record StopSiteAlarmAggregator;
|
||||
|
||||
/// <summary>Internal: a seed/reconcile snapshot fan-out completed with the whole-site alarm rows.</summary>
|
||||
internal sealed record SeedCompleted(IReadOnlyList<AlarmStateChanged> Alarms, bool IsInitial);
|
||||
|
||||
/// <summary>Internal: a seed/reconcile snapshot fan-out threw as a whole.</summary>
|
||||
internal sealed record SeedFailed(Exception Exception, bool IsInitial);
|
||||
|
||||
/// <summary>Internal: periodic reconcile tick (and the re-seed kicked after a reconnect).</summary>
|
||||
internal sealed record RunReconcile;
|
||||
|
||||
/// <summary>Internal: site-alarm gRPC stream error occurred.</summary>
|
||||
internal sealed record GrpcAlarmStreamError(Exception Exception);
|
||||
|
||||
/// <summary>Internal: reconnect the site-alarm gRPC stream (flip node).</summary>
|
||||
internal sealed record ReconnectAlarmStream;
|
||||
|
||||
/// <summary>Internal: the current site-alarm gRPC stream has stayed up long enough to recover the retry budget.</summary>
|
||||
internal sealed record GrpcAlarmStreamStable;
|
||||
@@ -88,4 +88,38 @@ public class CommunicationOptions
|
||||
/// that are deployed once and never re-deployed. Default 1 hour ≫ <see cref="PendingDeploymentTtl"/>.
|
||||
/// </summary>
|
||||
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.
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public TimeSpan LiveAlarmCacheLinger { get; set; } = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public TimeSpan LiveAlarmCacheReconcileInterval { get; set; } = TimeSpan.FromSeconds(60);
|
||||
|
||||
/// <summary>
|
||||
/// Max concurrent per-instance snapshot fetches during a live-cache seed/reconcile
|
||||
/// fan-out — the aggregated analogue of the Alarm Summary poll's
|
||||
/// <c>MaxConcurrentFetches</c>. Default 8.
|
||||
/// </summary>
|
||||
public int LiveAlarmCacheSeedConcurrency { get; set; } = 8;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public int LiveAlarmCacheMaxSubscribersPerSite { get; set; } = 200;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Communication;
|
||||
|
||||
/// <summary>
|
||||
/// 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 (<c>SubscribeSite</c>), so the page
|
||||
/// reflects alarm transitions in near-real-time instead of re-polling every 15s.
|
||||
/// <para>
|
||||
/// <b>Hard constraint (locked <c>[PERM]</c>):</b> 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Reference-counted lifecycle:</b> the first <see cref="Subscribe"/> for a site
|
||||
/// starts its aggregator (open stream + snapshot seed); the last subscriber leaving
|
||||
/// stops it after a short linger (see
|
||||
/// <see cref="CommunicationOptions.LiveAlarmCacheLinger"/>) to avoid re-seed thrash.
|
||||
/// One gRPC stream per <em>site</em>, not per browser circuit.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public interface ISiteAlarmLiveCache
|
||||
{
|
||||
/// <summary>
|
||||
/// Registers a viewer for a site's live alarm feed and returns an
|
||||
/// <see cref="IDisposable"/> that unregisters it. The first subscriber for a site
|
||||
/// starts the shared aggregator; the last subscriber's <see cref="IDisposable.Dispose"/>
|
||||
/// schedules a linger-delayed stop. <paramref name="onChanged"/> 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
|
||||
/// <c>IDeploymentStatusNotifier</c>). Disposing the returned handle is idempotent.
|
||||
/// </summary>
|
||||
/// <param name="siteId">The numeric site id whose alarms the viewer wants.</param>
|
||||
/// <param name="onChanged">Callback invoked whenever the site's cached alarm set changes.</param>
|
||||
/// <returns>A disposable that unregisters this viewer (idempotent).</returns>
|
||||
IDisposable Subscribe(int siteId, Action onChanged);
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
/// <param name="siteId">The numeric site id.</param>
|
||||
/// <returns>The current cached alarms (possibly empty), never <c>null</c>.</returns>
|
||||
IReadOnlyList<AlarmStateChanged> GetCurrentAlarms(int siteId);
|
||||
|
||||
/// <summary>
|
||||
/// 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): <c>false</c> means "not live yet — keep polling"; <c>true</c> means the
|
||||
/// cache is authoritative.
|
||||
/// </summary>
|
||||
/// <param name="siteId">The numeric site id.</param>
|
||||
/// <returns><c>true</c> once the site's aggregator has seeded and published at least once.</returns>
|
||||
bool IsLive(int siteId);
|
||||
}
|
||||
@@ -22,6 +22,13 @@ public static class ServiceCollectionExtensions
|
||||
services.AddSingleton<SiteStreamGrpcClientFactory>();
|
||||
services.AddSingleton<DebugStreamService>();
|
||||
|
||||
// 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<SiteAlarmLiveCacheService>();
|
||||
services.AddSingleton<ISiteAlarmLiveCache>(sp => sp.GetRequiredService<SiteAlarmLiveCacheService>());
|
||||
|
||||
// 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.
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Default <see cref="ISiteAlarmLiveCache"/> — 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
|
||||
/// <see cref="SiteAlarmAggregatorActor"/>, seeded by the existing debug-snapshot fan-out
|
||||
/// (<see cref="CommunicationService.RequestDebugSnapshotAsync"/>) 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.
|
||||
/// <para>
|
||||
/// <b>No persistence</b> (locked <c>[PERM]</c>): nothing here writes to the database — the
|
||||
/// cache lives only in the actor's memory on the active central node.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache
|
||||
{
|
||||
private static readonly IReadOnlyList<AlarmStateChanged> Empty = Array.Empty<AlarmStateChanged>();
|
||||
|
||||
private readonly IServiceProvider _serviceProvider;
|
||||
private readonly CommunicationService _communicationService;
|
||||
private readonly SiteStreamGrpcClientFactory _grpcFactory;
|
||||
private readonly CommunicationOptions _options;
|
||||
private readonly ILogger<SiteAlarmLiveCacheService> _logger;
|
||||
|
||||
private readonly object _lock = new();
|
||||
private readonly Dictionary<int, SiteEntry> _sites = new();
|
||||
private ActorSystem? _actorSystem;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes the live alarm cache service.
|
||||
/// </summary>
|
||||
/// <param name="serviceProvider">Root provider; scopes are created per snapshot fan-out to resolve repositories.</param>
|
||||
/// <param name="communicationService">Central-side comms used to Ask each instance for a debug snapshot (the seed).</param>
|
||||
/// <param name="grpcFactory">Factory for the per-site site-wide alarm gRPC client.</param>
|
||||
/// <param name="options">Communication options (linger, reconcile interval, seed concurrency, subscriber cap).</param>
|
||||
/// <param name="logger">Logger.</param>
|
||||
public SiteAlarmLiveCacheService(
|
||||
IServiceProvider serviceProvider,
|
||||
CommunicationService communicationService,
|
||||
SiteStreamGrpcClientFactory grpcFactory,
|
||||
IOptions<CommunicationOptions> options,
|
||||
ILogger<SiteAlarmLiveCacheService> logger)
|
||||
{
|
||||
_serviceProvider = serviceProvider;
|
||||
_communicationService = communicationService;
|
||||
_grpcFactory = grpcFactory;
|
||||
_options = options.Value;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Supplies the <see cref="ActorSystem"/> once it exists (called from
|
||||
/// <c>AkkaHostedService</c> on the active central node, mirroring
|
||||
/// <see cref="DebugStreamService.SetActorSystem"/>). Until set, <see cref="Subscribe"/>
|
||||
/// registers viewers but cannot start an aggregator, so the page keeps its poll fallback.
|
||||
/// </summary>
|
||||
/// <param name="actorSystem">The actor system to host per-site aggregators.</param>
|
||||
public void SetActorSystem(ActorSystem actorSystem) => _actorSystem = actorSystem;
|
||||
|
||||
/// <inheritdoc />
|
||||
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;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IReadOnlyList<AlarmStateChanged> GetCurrentAlarms(int siteId)
|
||||
{
|
||||
lock (_lock)
|
||||
return _sites.TryGetValue(siteId, out var entry) ? entry.Current : Empty;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Schedules a linger-delayed stop for a site whose last viewer just left. A
|
||||
/// re-subscribe within the linger window cancels it (<see cref="SiteEntry.CancelLinger"/>).
|
||||
/// Must be called under <see cref="_lock"/>.
|
||||
/// </summary>
|
||||
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<CancellationToken, Task<IReadOnlyList<AlarmStateChanged>>> seedFn =
|
||||
ct => FanOutSnapshotsAsync(entry.SiteId, ct);
|
||||
Action<IReadOnlyList<AlarmStateChanged>> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves a site's identifier + both gRPC node endpoints. Returns <c>null</c> (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.
|
||||
/// </summary>
|
||||
private async Task<(string SiteIdentifier, string GrpcA, string GrpcB)?> ResolveSiteAsync(int siteId)
|
||||
{
|
||||
using var scope = _serviceProvider.CreateScope();
|
||||
var siteRepo = scope.ServiceProvider.GetRequiredService<ISiteRepository>();
|
||||
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) ─────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Fans out one debug-snapshot Ask per Enabled instance on the site (capped at
|
||||
/// <see cref="CommunicationOptions.LiveAlarmCacheSeedConcurrency"/>) and returns every
|
||||
/// snapshot's alarm rows — <b>including</b> <see cref="AlarmStateChanged.IsConfiguredPlaceholder"/>
|
||||
/// placeholders. Best-effort: an instance that faults, times out, or reports
|
||||
/// <see cref="DebugViewSnapshot.InstanceNotFound"/> contributes no rows rather than
|
||||
/// failing the whole seed (mirrors <c>AlarmSummaryService</c>). Re-enumerates instances
|
||||
/// each call so a reconcile picks up enable/disable/deploy/delete drift.
|
||||
/// </summary>
|
||||
private async Task<IReadOnlyList<AlarmStateChanged>> FanOutSnapshotsAsync(int siteId, CancellationToken ct)
|
||||
{
|
||||
string siteIdentifier;
|
||||
List<string> enabledInstanceNames;
|
||||
|
||||
using (var scope = _serviceProvider.CreateScope())
|
||||
{
|
||||
var siteRepo = scope.ServiceProvider.GetRequiredService<ISiteRepository>();
|
||||
var site = await siteRepo.GetSiteByIdAsync(siteId, ct);
|
||||
if (site is null)
|
||||
return Empty;
|
||||
siteIdentifier = site.SiteIdentifier;
|
||||
|
||||
var instanceRepo = scope.ServiceProvider.GetRequiredService<ITemplateEngineRepository>();
|
||||
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<AlarmStateChanged>();
|
||||
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<AlarmStateChanged> 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<AlarmStateChanged> 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 ────────────────────────────────────────────────────────────
|
||||
|
||||
/// <summary>Per-site bookkeeping. All mutable fields are guarded by the service's <c>_lock</c>.</summary>
|
||||
private sealed class SiteEntry
|
||||
{
|
||||
public SiteEntry(int siteId) => SiteId = siteId;
|
||||
|
||||
public int SiteId { get; }
|
||||
public List<Subscription> Subscribers { get; } = new();
|
||||
public IActorRef? Actor { get; set; }
|
||||
public bool Starting { get; set; }
|
||||
|
||||
/// <summary>Current published snapshot the page renders. Swapped as a whole immutable list.</summary>
|
||||
public IReadOnlyList<AlarmStateChanged> Current { get; set; } = Empty;
|
||||
|
||||
/// <summary>True once the aggregator has seeded and published at least once.</summary>
|
||||
public bool HasPublished { get; set; }
|
||||
|
||||
public Timer? LingerTimer { get; set; }
|
||||
public int LingerVersion { get; set; }
|
||||
|
||||
/// <summary>Bounded retry timer armed when a start attempt failed to create the actor.</summary>
|
||||
public Timer? StartRetryTimer { get; set; }
|
||||
|
||||
/// <summary>Cancels any pending linger stop (a viewer arrived).</summary>
|
||||
public void CancelLinger()
|
||||
{
|
||||
LingerVersion++;
|
||||
LingerTimer?.Dispose();
|
||||
LingerTimer = null;
|
||||
}
|
||||
|
||||
/// <summary>Disposes both timers — called when the entry is torn down.</summary>
|
||||
public void DisposeTimers()
|
||||
{
|
||||
LingerTimer?.Dispose();
|
||||
LingerTimer = null;
|
||||
StartRetryTimer?.Dispose();
|
||||
StartRetryTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The disposable handed to a viewer; idempotent <see cref="Dispose"/> unregisters it.</summary>
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -424,6 +424,11 @@ akka {{
|
||||
var debugStreamService = _serviceProvider.GetService<DebugStreamService>();
|
||||
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<ZB.MOM.WW.ScadaBridge.Communication.SiteAlarmLiveCacheService>();
|
||||
siteAlarmLiveCache?.SetActorSystem(_actorSystem!);
|
||||
|
||||
|
||||
// Management Service — accessible via ClusterClient
|
||||
var mgmtLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
|
||||
|
||||
+414
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for <see cref="SiteAlarmAggregatorActor"/> (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).
|
||||
/// </summary>
|
||||
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 ────────────────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
private sealed class SeedStub
|
||||
{
|
||||
private readonly ConcurrentQueue<TaskCompletionSource<IReadOnlyList<AlarmStateChanged>>> _pending = new();
|
||||
private int _callCount;
|
||||
|
||||
public int CallCount => Volatile.Read(ref _callCount);
|
||||
|
||||
public Task<IReadOnlyList<AlarmStateChanged>> Seed(CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref _callCount);
|
||||
var tcs = new TaskCompletionSource<IReadOnlyList<AlarmStateChanged>>(
|
||||
TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
ct.Register(() => tcs.TrySetCanceled());
|
||||
_pending.Enqueue(tcs);
|
||||
return tcs.Task;
|
||||
}
|
||||
|
||||
/// <summary>Completes the oldest not-yet-completed seed call with the given rows.</summary>
|
||||
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.");
|
||||
}
|
||||
|
||||
/// <summary>Faults the oldest not-yet-completed seed call (simulates a whole-fan-out failure).</summary>
|
||||
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<IReadOnlyList<AlarmStateChanged>> Snapshots { get; } = new();
|
||||
|
||||
public void Publish(IReadOnlyList<AlarmStateChanged> snapshot)
|
||||
{
|
||||
lock (_lock) { Snapshots.Add(snapshot); }
|
||||
}
|
||||
|
||||
public IReadOnlyList<AlarmStateChanged>? 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<AlarmStateChanged> OnAlarm, Action<Exception> OnError, CancellationToken Ct);
|
||||
|
||||
private sealed class MockSiteAlarmStreamClient : SiteStreamGrpcClient
|
||||
{
|
||||
private readonly object _lock = new();
|
||||
private readonly List<SiteSub> _subs = new();
|
||||
private readonly List<string> _unsubscribed = new();
|
||||
|
||||
public List<SiteSub> Subs { get { lock (_lock) { return _subs.ToList(); } } }
|
||||
public List<string> Unsubscribed { get { lock (_lock) { return _unsubscribed.ToList(); } } }
|
||||
|
||||
public MockSiteAlarmStreamClient() : base() { }
|
||||
|
||||
public override Task SubscribeSiteAsync(
|
||||
string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> 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<string, MockSiteAlarmStreamClient> _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<AlarmStateChanged>? 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));
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for <see cref="SiteAlarmLiveCacheService"/> (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.
|
||||
/// </summary>
|
||||
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<Commons.Messages.Streaming.AlarmStateChanged> onAlarmEvent,
|
||||
Action<Exception> 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<ISiteRepository>();
|
||||
siteRepo.GetSiteByIdAsync(SiteId, Arg.Any<CancellationToken>()).Returns(site);
|
||||
|
||||
var instanceRepo = Substitute.For<ITemplateEngineRepository>();
|
||||
instanceRepo.GetInstancesBySiteIdAsync(SiteId, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<Instance>());
|
||||
|
||||
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<CommunicationService>.Instance);
|
||||
|
||||
factory = new CountingFactory();
|
||||
var service = new SiteAlarmLiveCacheService(
|
||||
provider, comm, factory, options, NullLogger<SiteAlarmLiveCacheService>.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<ISiteRepository>();
|
||||
// First resolve returns null (transient blip); every resolve thereafter succeeds.
|
||||
siteRepo.GetSiteByIdAsync(SiteId, Arg.Any<CancellationToken>()).Returns((Site?)null, site);
|
||||
|
||||
var instanceRepo = Substitute.For<ITemplateEngineRepository>();
|
||||
instanceRepo.GetInstancesBySiteIdAsync(SiteId, Arg.Any<CancellationToken>())
|
||||
.Returns(new List<Instance>());
|
||||
|
||||
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<CommunicationService>.Instance);
|
||||
var factory = new CountingFactory();
|
||||
|
||||
var service = new SiteAlarmLiveCacheService(
|
||||
provider, comm, factory, options, NullLogger<SiteAlarmLiveCacheService>.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));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user