docs(comm): accept + document standby aggregators and deleted-site viewer behavior; fix aggregator e2e ctor call for new params (plan R2-02 T14)

This commit is contained in:
Joseph Doherty
2026-07-13 10:25:10 -04:00
parent 4e228110ec
commit 41ffe42de3
3 changed files with 12 additions and 5 deletions
+4 -1
View File
@@ -75,8 +75,11 @@ Delivered 2026-07-10 (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.m
- **Central live cache** (`ISiteAlarmLiveCache`, singleton `SiteAlarmLiveCacheService`): a DI singleton on the active central node. For each site with ≥1 active viewer it runs ONE shared, **reference-counted** per-site aggregator (`SiteAlarmAggregatorActor`); the first `Subscribe(siteId, onChanged)` starts it, the last subscriber leaving stops it after a short **linger** to avoid re-seed thrash. `GetCurrentAlarms(siteId)` returns the current immutable snapshot; `IsLive(siteId)` reports whether the aggregator has seeded and published at least once. - **Central live cache** (`ISiteAlarmLiveCache`, singleton `SiteAlarmLiveCacheService`): a DI singleton on the active central node. For each site with ≥1 active viewer it runs ONE shared, **reference-counted** per-site aggregator (`SiteAlarmAggregatorActor`); the first `Subscribe(siteId, onChanged)` starts it, the last subscriber leaving stops it after a short **linger** to avoid re-seed thrash. `GetCurrentAlarms(siteId)` returns the current immutable snapshot; `IsLive(siteId)` reports whether the aggregator has seeded and published at least once.
- **Seed-then-stream** (copied from `DebugStreamBridgeActor` ordering): open the `SubscribeSite` stream first (buffer live deltas), run the snapshot fan-out once via the existing `DebugViewSnapshot` path (bounded by `LiveAlarmCacheSeedConcurrency`), flush the buffer with **dedup by `(InstanceUniqueName, AlarmName, SourceReference)`**, then live pass-through. Placeholders are seeded from the snapshot and never expected on the live stream. - **Seed-then-stream** (copied from `DebugStreamBridgeActor` ordering): open the `SubscribeSite` stream first (buffer live deltas), run the snapshot fan-out once via the existing `DebugViewSnapshot` path (bounded by `LiveAlarmCacheSeedConcurrency`), flush the buffer with **dedup by `(InstanceUniqueName, AlarmName, SourceReference)`**, then live pass-through. Placeholders are seeded from the snapshot and never expected on the live stream.
- **Failover & drift**: NodeA↔NodeB reconnect (client via `SiteStreamGrpcClient.SubscribeSiteAsync`, factory endpoint failover) re-seeds rather than serving stale; a periodic **reconcile snapshot** (default 60s) corrects any missed enable/disable so the cache can't drift indefinitely. `[PERM]` (`docs/plans/2026-05-29-native-alarms-design.md`): the cache is **purely in-memory** — no EF entity/table/migration, no persisted central alarm store — so a new active node simply re-seeds from scratch. - **Failover & drift**: NodeA↔NodeB reconnect (client via `SiteStreamGrpcClient.SubscribeSiteAsync`, factory endpoint failover) re-seeds rather than serving stale; a periodic **reconcile snapshot** (default 60s) corrects any missed enable/disable so the cache can't drift indefinitely. `[PERM]` (`docs/plans/2026-05-29-native-alarms-design.md`): the cache is **purely in-memory** — no EF entity/table/migration, no persisted central alarm store — so a new active node simply re-seeds from scratch.
- **Options** (`Communication` section, `CommunicationOptions`; eagerly validated by `CommunicationOptionsValidator` / `ValidateOnStart`): `LiveAlarmCacheLinger` (default 30s), `LiveAlarmCacheReconcileInterval` (default 60s), `LiveAlarmCacheSeedConcurrency` (default 8), `LiveAlarmCacheMaxSubscribersPerSite` (default 200). - **Options** (`Communication` section, `CommunicationOptions`; eagerly validated by `CommunicationOptionsValidator` / `ValidateOnStart`): `LiveAlarmCacheLinger` (default 30s), `LiveAlarmCacheReconcileInterval` (default 60s), `LiveAlarmCacheSeedConcurrency` (default 8), `LiveAlarmCacheMaxSubscribersPerSite` (default 200), `LiveAlarmCachePublishCoalesce` (default 250ms; `0` = publish per delta — legacy — batches an alarm storm into one snapshot copy + one viewer fan-out per window; arch review 02 round 2, N6).
- **Telemetry** (`ScadaBridgeTelemetry` meter): observable gauge `scadabridge.site.alarm_cache.aggregators.active` (running per-site aggregators) and counter `scadabridge.site.alarm_cache.reconnects` (site-wide stream reconnects — a NodeA↔NodeB flip or reconcile-driven reopen; a sustained climb signals a flapping site link). - **Telemetry** (`ScadaBridgeTelemetry` meter): observable gauge `scadabridge.site.alarm_cache.aggregators.active` (running per-site aggregators) and counter `scadabridge.site.alarm_cache.reconnects` (site-wide stream reconnects — a NodeA↔NodeB flip or reconcile-driven reopen; a sustained climb signals a flapping site link).
- **Accepted limitations (arch review 02 round 2, N8):**
- **Standby-node aggregators**: the live cache is per-node and `SetActorSystem` is wired on every central node, so browsing the standby node directly (diagnostic ports, e.g. 9002) starts a second, fully-functional read-only aggregator + `SubscribeSite` stream there. Accepted — not gated: the aggregator is read-only, bounded (one stream/site, viewer-capped), and torn down by the viewer linger; gating `SetActorSystem` behind the active check would break the diagnostic-browse path and buy nothing. The `[PERM]` "lives only on the active central node" claim is corrected to "per-node; in routine operation only the active node hosts viewers".
- **Site deleted while an Alarm Summary viewer is open**: the viewer's aggregator reconciles to an empty snapshot (the deleted site's instances vanish from the fan-out) until its viewers leave, then the linger stop reaps it; the site's cached gRPC channels are disposed at deletion via `SiteStreamGrpcClientFactory.RemoveSiteAsync` (see `ManagementActor.HandleDeleteSite`, arch review 02 round 2, N8).
#### Site-Side gRPC Streaming Components #### Site-Side gRPC Streaming Components
@@ -11,9 +11,12 @@ namespace ZB.MOM.WW.ScadaBridge.Communication;
/// reflects alarm transitions in near-real-time instead of re-polling every 15s. /// reflects alarm transitions in near-real-time instead of re-polling every 15s.
/// <para> /// <para>
/// <b>Hard constraint (locked <c>[PERM]</c>):</b> there is NO persisted central alarm /// <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 — /// store. This cache is purely in-memory, per-node — no EF entity/table/migration backs
/// no EF entity/table/migration backs it. On a NodeA↔NodeB failover the new active /// it. In routine operation only the active node (behind Traefik) hosts viewers, so only
/// node re-seeds from scratch. /// it runs aggregators; browsing the standby node directly (diagnostic ports) starts an
/// independent read-only aggregator there — accepted (arch review 02 round 2, N8):
/// bounded (one stream/site, viewer-capped), read-only, and torn down by the viewer
/// linger. On a NodeA↔NodeB failover the new active node re-seeds from scratch.
/// </para> /// </para>
/// <para> /// <para>
/// <b>Reference-counted lifecycle:</b> the first <see cref="Subscribe"/> for a site /// <b>Reference-counted lifecycle:</b> the first <see cref="Subscribe"/> for a site
@@ -223,7 +223,8 @@ public class SiteAlarmStreamEndToEndTests : TestKit
var sink = new PublishSink(); var sink = new PublishSink();
var aggregator = Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor( var aggregator = Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor(
"site-alpha", "e2e-parity", _ => Task.FromResult<IReadOnlyList<AlarmStateChanged>>(new[] { seedRow }), "site-alpha", "e2e-parity", _ => Task.FromResult<IReadOnlyList<AlarmStateChanged>>(new[] { seedRow }),
sink.Publish, factory, "http://a:5100", "http://b:5100", TimeSpan.FromMinutes(10)))); sink.Publish, factory, "http://a:5100", "http://b:5100", TimeSpan.FromMinutes(10),
TimeSpan.Zero, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60))));
await WaitForConditionAsync(() => sink.Latest is { Count: 1 }); // seed applied await WaitForConditionAsync(() => sink.Latest is { Count: 1 }); // seed applied
aggregator.Tell(liveRow); aggregator.Tell(liveRow);