ScadaBridge/archreview/02-communication-store-and-forward.md

Architecture Review — Central–Site Communication & Store-and-Forward (Round 2)

Reviewer domain: src/ZB.MOM.WW.ScadaBridge.Communication, src/ZB.MOM.WW.ScadaBridge.StoreAndForward, the telemetry transport paths that ride them, the Host wiring that composes them, and — new this round — the aggregated live alarm stream shipped 2026-07-10 (docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.md): site-wide alarm-only SubscribeSite gRPC stream, central ISiteAlarmLiveCache/SiteAlarmLiveCacheService/SiteAlarmAggregatorActor, seed-then-stream, and reconnect telemetry.

Date: 2026-07-12 (round 2). Round-1 report dated 2026-07-08 at baseline b910f5eb; this round re-reviews everything landed since (PLAN-02's 24 tasks, PLAN-08 options validation, and the post-initiative live-alarm-stream feature) on main @ 8c888f13.

Scope & Method

• Read the round-1 report and archreview/plans/PLAN-02-communication-store-and-forward.md (findings-coverage table), then verified every round-1 finding against the current source — no plan claim was trusted without file:line evidence.
• Fresh sweep over git diff b910f5eb..HEAD scoped to the Communication and StoreAndForward projects and their tests (~4,900 insertions across 50 files, 32 commits), plus the SiteRuntime/Host/CentralUI files the new code touches (SiteReplicationActor, SiteStreamManager, AkkaHostedService, ScriptRuntimeContext, AlarmSummary.razor).
• Static review only (suites verified green 2026-07-10). Known context: the SiteStreamGrpcServerTests.SiteConnectionUpGauge_* observable-gauge timing flake pre-exists on main and is not a regression from these plans.

One-line verdict: round 1's Critical and all three Highs are genuinely fixed and well-tested — but the new anti-entropy resync was built on the wrong active-node predicate (leader instead of the oldest-Up helper the delivery gate uses) and can wipe the delivering node's live buffer after a routine rolling restart, and its snapshot message cannot fit through Akka remoting's default frame size — so round 2 finds one new Critical and one new High, both confined to the peer-resync feature; everything else, including the brand-new live alarm stream, is solid work.

---

Round-1 Finding Disposition

Round-1 finding	R1 severity	Disposition	Evidence (file:line)
ClusterClient recreate-on-address-change → name collision → restart loop	Critical	Fixed (verified)	CentralCommunicationActor.cs:33-67 — DefaultSiteClientFactory mints site-client-{sanitized}-{generation} via Interlocked.Increment (48); real-factory lifecycle tests exist (tests/.../DefaultSiteClientFactoryTests.cs, CentralCommunicationActorClientLifecycleTests.cs)
siteId interpolated unvalidated into actor name (sub-finding)	Critical (sub)	Fixed (verified)	CentralCommunicationActor.cs:59-66 — SanitizeForActorName maps to valid path elements, falls back to "site"; Create additionally wrapped in try/catch that logs and skips the site (582-593), stale entry removed before recreate (572-580)
Standby site node runs full S&F delivery pipeline (systematic duplicates)	High	Fixed (verified)	StoreAndForwardService.cs:89-103 (gate seam), 682-698 (re-evaluated per sweep tick, throwing gate = standby); Host wires it to the shared singleton-host helper IClusterNodeProvider.SelfIsPrimary (AkkaHostedService.cs:868-884); unit tests cover standby-skip / flip-active / throwing-gate. But see NEW finding N2 — the resync feature added alongside uses a different predicate
Debug-session gRPC failover disposes shared channel; CleanupGrpc creates channels during teardown	High	Fixed (verified)	SiteStreamGrpcClientFactory.cs:26,61-73 — cache keyed by (site, endpoint), GetOrCreate never disposes, TryGet never creates; RemoveSiteAsync (95-102) is the only shared-disposal path; DebugStreamBridgeActor.cs:469-473, 495-498 — both teardown paths use TryGet(...)?.Unsubscribe(...)
Retry sweep serial, unbounded, single-laned — collapses under backlog	High (perf)	Fixed (verified)	StoreAndForwardOptions.cs:31 SweepBatchLimit (500) → StoreAndForwardStorage.cs:392-397 SQL LIMIT; per-(category,target) lane short-circuit on first transient failure (StoreAndForwardService.cs:721-730, RetryOutcome 752); parallel lanes capped at SweepTargetParallelism 4, sequential within lane (710-734)
Replication ops applied out of order (Task.Run fire-and-forget)	Medium	Fixed (verified)	ReplicationService.cs:142-161 — handler invoked inline; ordering preserved by construction, faults observed via ContinueWith(OnlyOnFaulted)
Replication failures logged at Debug — dead channel invisible	Medium	Fixed (verified)	ReplicationService.cs:153-168 — Warning + ScadaBridgeTelemetry.RecordReplicationFailure(); SiteReplicationActor.cs:225-234 — no-peer drop also Warning + counter
Park/Requeue replication blind UPDATE — lost on standby missing the row	Medium	Fixed (verified)	ReplicationService.cs:115-140 — Add/Park/Requeue applied via StoreAndForwardStorage.UpsertMessageAsync (ON CONFLICT(id) DO UPDATE, StoreAndForwardStorage.cs:319-369, created_at preserved)
Notifications park after ~25 min; maxRetries: 0 escape hatch unused	Medium	Fixed (verified)	ScriptRuntimeContext.cs:2209 — Notify.Send enqueues with maxRetries: 0; park-on-exhaustion is impossible (StoreAndForwardService.cs:830 guards MaxRetries > 0)
Corrupt buffered notification discarded and reported as delivered	Medium	Fixed (verified)	NotificationForwarder.cs:82-95 — undeserialisable payload returns false → engine parks the row; Warning log with capped payload preview (170-187)
Spec-vs-code drift: standby-passive model; Notify.Send blocks up to 30 s	Medium	Fixed (verified)	Component-StoreAndForward.md:80-83 — gate + failover-window-only duplicates documented; :52 — enqueue-only deferToSweep semantics documented; code matches (StoreAndForwardService.cs:588-594 deferToSweep path never invokes the handler inline, kicks TriggerSweep)
No explicit Akka serializer — wire format unpinned	Medium (perf)	Partially fixed / deferred (accepted)	Round-trip pin tests added (tests/.../WireSerializationPinTests.cs — 191 lines, ReplicationWireSerializationPinTests.cs; PLAN-08 T9 additionally found the Newtonsoft ctor-optional-default gotcha and locked contracts). The serializer swap remains deferred by design (rolling-compat migration) — reasonable
Central heartbeat state not replicated to peer central node	Low	Fixed (verified)	CentralCommunicationActor.cs:369-398 — SiteHeartbeatReplica fanned over DistributedPubSub, idempotent local mark (194-196, 390-398)
SubscribeInstance concurrency check-then-act (comment only asked)	Low	Fixed as specified	SiteStreamGrpcServer.cs:287-292 — deliberate check-then-act documented with bounded-overshoot rationale
S&F SQLite: no WAL, no busy_timeout	Low	Fixed (verified)	StoreAndForwardStorage.cs:43-47 (WAL in InitializeAsync), 177-185 (5 s busy_timeout per pooled open)
Due-check per-row julianday() parsing	Low	Fixed (verified)	Additive last_attempt_at_ms column + one-time backfill + (status, last_attempt_at_ms) index (StoreAndForwardStorage.cs:88-114); due predicate is pure integer math (386-400)
Audit-observer notification awaited inline in the sweep	Low	Fixed (verified)	StoreAndForwardService.cs:117-135, 898-922 — ordered single-reader channel + pump; inline fallback preserved when not started; drained on StopAsync (451-468)
StreamRelayActor.WriteToChannel dead branch + silent DropOldest loss	Low	Fixed (verified)	StreamRelayActor.cs:105-115 — comment corrected (guards closed channel only); real eviction counted by the bounded channel's itemDropped callback with first + every-500th Warning (SiteStreamGrpcServer.cs:303-315)
Duplicated Ask-timeout constants between the two ingest transports	Low	Fixed (verified)	Single source of truth SiteStreamGrpcServer.AuditIngestAskTimeout (SiteStreamGrpcServer.cs:47), consumed by CentralCommunicationActor.cs:140-150, 174
Heartbeat IsActive leader-check vs oldest-node singleton placement	Low	Partially fixed — and now materially worse	PLAN-01 shipped the oldest-Up helper (ClusterActivityEvaluator.SelfIsOldest, Host/Health/ClusterActivityEvaluator.cs:20-31; AkkaClusterNodeProvider.SelfIsPrimary:29-37) and the S&F delivery gate uses it — but SiteCommunicationActor.DefaultIsActiveCheck (517-526) and SiteReplicationActor.DefaultIsActive (190-198) still use leader+Up with "swap point" comments never swapped. The new resync feature was built on the leader predicate → NEW finding N2 (Critical)
TransportHeartbeatInterval double duty	Low	Fixed (verified)	CommunicationOptions.cs:59-65 — dedicated ApplicationHeartbeatInterval (default 5 s, no behavior change), used at SiteCommunicationActor.cs:444; validated (CommunicationOptionsValidator.cs:54-55)
U1: Dual transports for the same ingest operations	Underdev	Deferred (accepted)	Both paths still exist (SiteStreamGrpcServer.cs:388-491 gRPC vs CentralCommunicationActor.cs:282-325 ClusterClient) with the shared timeout constant removing the tuning drift; consolidation remains a registered follow-up. Acceptable
U2: No real-collaborator test for ClusterClient cache lifecycle	Underdev	Fixed (verified)	DefaultSiteClientFactoryTests.cs (real factory, collision + sanitize), CentralCommunicationActorClientLifecycleTests.cs (address-edit ×3, throwing factory survives)
U3: No test/design treatment of standby-node sweep behavior	Underdev	Fixed (unit level)	Gate tests in StoreAndForwardServiceTests.cs (skip/resume/throwing-gate); doc updated; the live two-node kill rig landed in PLAN-01 (TwoNodeClusterFixture)
U4: Replication has no anti-entropy/resync	Underdev	Fixed — with two new defects	SiteReplicationActor.cs:111-113, 175-181, 398-444 + StoreAndForwardStorage.GetAllMessagesAsync/ReplaceAllAsync (256-306). The mechanism exists and is authority-checked and capped — but see N1 (frame size) and N2 (predicate inversion) below
U5: Parked rows unbounded retention, no aging signal	Underdev	Fixed (verified)	StoreAndForwardStorage.GetOldestParkedCreatedAtAsync (675-685) → HealthReportSender.cs:114-115 → additive SiteHealthReport.OldestParkedMessageAgeSeconds (SiteHealthReport.cs:57)
U6: Reconciliation cursor edge (> vs >=) unpinned	Underdev	Fixed (verified, exceeded)	PullSiteCalls grew an additive composite-keyset cursor (after_id, SiteStreamGrpcServer.cs:598-641); pinned by SiteStreamPullSiteCallsTests.cs
U7: DebugStreamService._sessions lost silently on central restart	Underdev	Documented (accepted)	Component-Communication.md:67 — explicit "Accepted limitation" paragraph citing this review

Disposition counts: 22 Fixed (verified) · 2 Partially fixed (serializer pin-only by design; leader-vs-oldest helper shipped but not swapped into all call sites) · 2 Deferred/accepted (U1 dual transports, U7 session locality) · 0 Not fixed · 0 Regressed. Round-1 Critical C2/C1 (ClusterClient recreate) specially verified: genuinely fixed, with real-collaborator tests exactly where round 1 said the blind spot was.

---

NEW Findings — Store-and-Forward & Replication

[Critical] N1 — Anti-entropy resync authority uses the leader predicate while delivery uses oldest-Up: after a routine rolling restart the stale rejoining node wipes the delivering node's live buffer

The delivery gate installed by PLAN-02 T4 correctly uses PLAN-01's singleton-host helper: AkkaHostedService.cs:881 → AkkaClusterNodeProvider.SelfIsPrimary → ClusterActivityEvaluator.SelfIsOldest (ClusterActivityEvaluator.cs:20-31, up-number/IsOlderThan based — matches where cluster singletons actually run). But the peer-join resync roles added in the same plan use a different predicate: SiteReplicationActor.DefaultIsActive (SiteReplicationActor.cs:190-198) is leader + Up, with a comment calling itself the "swap point for plan 01's shared helper" that was never swapped.

Akka.NET's cluster leader is the lowest-address reachable Up member — not the oldest. The two predicates diverge persistently (not just transiently) whenever the lower-address node is not the oldest, which is exactly the state produced by the most routine operation there is: a rolling restart (or crash + rejoin) of the lower-address site node. Sequence, with node-a (lower address) restarted while node-b keeps running:

1. node-b is now oldest → SelfIsPrimary true → node-b runs the delivery sweep and owns the live buffer. node-a rejoins with a fresh (higher) up-number but becomes leader by address ordering.
2. node-b observes MemberUp(node-a) → TryTrackPeer → OnPeerTracked (SiteReplicationActor.cs:175-181): SafeIsActive() is false on node-b (it is not leader) → the delivering node sends RequestSfBufferResync to the stale peer.
3. node-a handles it (HandleRequestSfBufferResync, 398-411): SafeIsActive() is true on node-a (leader+Up) → it answers with a snapshot of its stale, pre-outage buffer.
4. node-b handles the snapshot (HandleSfBufferSnapshot, 420-444): SafeIsActive() false → it calls ReplaceAllAsync — the method whose own doc says "Never call on an active node — it discards every in-flight row" (StoreAndForwardStorage.cs:281-283) — on the node that is actively delivering.

Consequences: every message buffered on node-b during and since node-a's outage is deleted (notifications enqueued with maxRetries: 0 are silently gone — the one category the design promises is never lost short of payload corruption), and rows node-b already delivered are resurrected from node-a's stale copy → guaranteed duplicate deliveries on subsequent sweeps. ReplicationEnabled defaults to true (StoreAndForwardOptions.cs:12), so production 2-node sites are exposed by default.

Fix shape: swap both remaining leader-based predicates (SiteReplicationActor.cs:190-198, SiteCommunicationActor.cs:517-526) to the shared oldest-Up helper so one definition of "active" governs delivery, resync-request, resync-answer, and heartbeat IsActive; add a two-node rig test that rolls the lower-address node and asserts the surviving node's buffer is untouched. (Also consider a belt-and-braces guard: refuse ReplaceAllAsync while this node's delivery gate reports active.)

[High] N2 — The resync snapshot rides ONE Akka remoting message with no frame-size headroom: it is silently undeliverable for any realistic backlog

HandleRequestSfBufferResync pipes up to 10,000 full rows (id + target + complete payload_json + provenance) into a single SfBufferSnapshot record and Tells it across Akka remoting (SiteReplicationActor.cs:40, 407-410, 460). BuildHocon sets no akka.remote.dot-netty.tcp.maximum-frame-size (grep across src/ and docker/ finds none), so the default 128,000 bytes applies. That is ~12 bytes per row at the cap — even a few hundred typical cached-call payloads exceed it. An oversized remote message is dropped by the transport (an error in the sender's log; no exception surfaces to the PipeTo), the standby never receives the snapshot, nothing retries until the next peer-track event, and no telemetry counts it. Net effect: the anti-entropy fix works only for near-empty buffers and silently fails in exactly the sustained-outage/backlog scenario it was built for (U4's "standby down for an hour"). Ironically this also caps N1's blast radius — large stale snapshots can't be delivered either — but small-buffer wipes still go through.

Fix shape: chunk the snapshot (reuse the pull-style paging the audit reconciliation already has), or stream rows as a sequence of upsert ops bracketed by begin/end markers with a completion-driven delete-absent pass; count send failures. Also reconsider the 10k in-memory List on the active node.

[Medium] N3 — _sweepTask handoff is clobbered by every timer tick, defeating the shutdown drain it exists for

Both the timer callback (StoreAndForwardService.cs:396-400) and TriggerSweep (664-668) unconditionally Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync()). When a sweep is still in flight, the new call no-ops via the _retryInProgress CAS and returns a completed task — which overwrites _sweepTask. StopAsync (422-431) then awaits the no-op and proceeds with shutdown while the real sweep (one that has outlived a 10 s tick — i.e. precisely the backlog case the drain was added for) is still mid-delivery. Consequence is bounded (the sweep's own catch logs; storage is connection-per-op), but the "wait for the in-flight sweep" guarantee is illusory whenever it matters. Fix: only publish the task when the CAS is won (e.g. have RetryPendingMessagesAsync register itself), or CAS _sweepTask from completed→new.

[Low] N4 — New sweep options missing from the eager validator

StoreAndForwardOptionsValidator (17-34) validates path/intervals/max-retries but not the two options PLAN-02 added: a negative SweepBatchLimit silently means "unlimited (legacy)" (StoreAndForwardStorage.cs:392 checks > 0) and a non-positive SweepTargetParallelism is silently clamped to 1 (StoreAndForwardService.cs:715). Both deserve fail-fast messages per the §1.5 eager-validation convention the same release adopted elsewhere.

[Low] N5 — Resync snapshot vs in-flight replication op race leaves orphan rows on the standby

A replicated Remove sent after the active node read its snapshot but before the snapshot send is ordered before the snapshot on the wire (same sender/receiver pair), so the stale snapshot re-adds the removed row on the standby → an orphan Pending row that a later failover re-delivers. Bounded and self-correcting at the next resync, and inherent to no-ack replication — worth a code comment on HandleSfBufferSnapshot (SiteReplicationActor.cs:420-444) so nobody "fixes" it into something worse.

---

NEW Findings — Aggregated Live Alarm Stream (post-initiative, first review)

The feature is well-shaped overall: one gRPC stream per site (not per circuit), reference-counted lifecycle with linger, stream-first seed with ordered delta buffering and per-key dedup (correctly copied from DebugStreamBridgeActor, including the inclusive-boundary rule), authority-split with the poll fallback (IsLive false until first seed publishes — SiteAlarmLiveCacheService.cs:131-136, 306-323), reconcile as the drop-detection backstop (SiteAlarmAggregatorActor.cs:276-304 rebuilds authoritatively so disappeared rows are dropped), self-healing on both start failure (bounded retry timer, SiteAlarmLiveCacheService.cs:249-275) and stream give-up (reconcile tick reopens, SiteAlarmAggregatorActor.cs:224-236), balanced telemetry gauges in PreStart/PostStop (188, 212), viewer-cap fail-safe returning a no-op handle instead of throwing on a Blazor render path (100-107), and eager options validation (CommunicationOptionsValidator.cs:69-87). Findings:

[Medium] N6 — No delta coalescing: every applied alarm delta copies the whole site cache and fans out to every viewer circuit

HandleLiveDelta → Publish() materialises a fresh _cache.Values.ToList() per delta (SiteAlarmAggregatorActor.cs:365-367, 391-403), and OnPublish invokes every subscriber's onChanged (SiteAlarmLiveCacheService.cs:386-411), each of which does a Blazor InvokeAsync + GetCurrentAlarms + re-render (AlarmSummary.razor:369-388). An alarm flood — the canonical SCADA storm scenario, hundreds of condition transitions per second on a large site — becomes O(cacheSize) list copies per transition on the actor thread plus a render queue entry per circuit per transition. The seed path already batches; the live path should too (e.g. a ~250 ms publish-coalescing timer, or publish-on-tick-if-dirty). Everything stays correct (last write wins), so this is performance, not correctness.

[Low] N7 — Aggregator lifecycle nits (bundled)

1. Reconnect re-seed can be skipped: the post-failover-flip StartFanout(isInitial: false) (SiteAlarmAggregatorActor.cs:472) no-ops if a reconcile fan-out is already in flight (247-253) — deltas missed between stream death and the in-flight snapshot's read-time stay stale until the next reconcile (up to 60 s), despite the "never silently serve stale" comment.
2. No stream-generation stamp on GrpcAlarmStreamError: a late error raced out of the previous cancelled stream (the RpcException(Cancelled) filter at SiteStreamGrpcClient.cs:256-258 covers the normal path, but a genuine socket fault can beat the cancel) is indistinguishable from a failure of the current stream and burns retry budget / double-flips. The window is small; a monotonic stream-id in the error message would close it.
3. Static mutable test seams: ReconnectDelay / StabilityWindow are internal static mutable properties (SiteAlarmAggregatorActor.cs:61-69) — process-global state that parallel test classes can trample; instance/ctor parameters would match how the rest of the file injects tuning.

[Low] N8 — "Lives only on the active central node" is aspirational, and deleted sites leak idle channels

SetActorSystem is wired on every central node (AkkaHostedService.cs:430), so browsing the standby node directly (port 9002, documented for diagnostics) starts a second, fully functional aggregator + gRPC stream there — harmless (read-only) but contradicting the [PERM] doc claim in ISiteAlarmLiveCache.cs:13-17. Separately, SiteStreamGrpcClientFactory.RemoveSiteAsync — designed as "the disposal path when a site record is deleted" (SiteStreamGrpcClientFactory.cs:88-102) — has no production caller (grep: tests only), so a deleted site's channels (and an aggregator kept alive by an open viewer, which then reconciles to an empty snapshot forever) persist until process restart. Bounded, but wire the deletion path or document the gap.

[Low] N9 — A site with only one configured gRPC endpoint can never go live

ResolveSiteAsync requires both GrpcNodeAAddress and GrpcNodeBAddress (SiteAlarmLiveCacheService.cs:294-300); a single-node site (or one with only NodeA populated) silently never starts an aggregator and polls forever, with only a log Warning. The aggregator itself would work fine flipping between one address; accept a single endpoint and disable the flip.

---

What's genuinely good (round 2)

• The fix quality is high and test-backed. Every round-1 fix landed with the failing-test-first shape the plan prescribed; the two places round 1 called out as structurally untested (real DefaultSiteClientFactory, standby sweep behavior) now have real-collaborator tests. The wire-format pin tests even caught a real Newtonsoft deserialization gotcha (ctor optional-param defaults) during PLAN-08.
• The sweep redesign is careful. Skipped lane rows keep RetryCount/LastAttemptAt untouched so the park horizon reflects real attempts; GroupBy stability preserves per-target FIFO; the observer pump preserves per-operation event order while latency-isolating the sweep; the gate is fail-safe (throwing → standby) and re-evaluated per tick.
• Storage hygiene done right: additive last_attempt_at_ms with one-time guarded backfill and a covering (status, due) index; shared InsertMessageSql + BindMessageParameters so column lists can't drift; defensive Guid.TryParse on read so one corrupt row can't abort a sweep.
• The live alarm stream reuses instead of reinventing: SubscribeSite rides the same hardened RunSubscriptionStreamAsync pipeline (readiness/shutdown gates, correlation-id validation, duplicate replacement, DropOldest with counted evictions, leak-proof subscribe failure cleanup — SiteStreamGrpcServer.cs:253-385); the client shares the proto→domain mapping; the actor copies the proven seed-then-stream/buffer/dedup/stability-window design from DebugStreamBridgeActor including its NUL-delimited three-part alarm key; the proto change is purely additive (new RPC + new request message).
• Failure-mode thinking is visible everywhere in the new code: seed fan-out degrades per-instance rather than failing whole; a failed initial seed keeps IsLive false so the page keeps polling; publish callbacks are exception-isolated outside the lock; the viewer cap rejects with a no-op disposable rather than throwing into a render.
• Docs were genuinely synchronized: Component-StoreAndForward.md (gate, resync, enqueue-only Send) and Component-Communication.md (U7 accepted-limitation) now match the code they describe.

---

Severity tally — NEW findings only

Severity	Count	Findings
Critical	1	N1 (resync predicate inversion wipes the delivering node's buffer)
High	1	N2 (resync snapshot exceeds Akka remoting frame size — silently undeliverable)
Medium	2	N3 (_sweepTask clobbering defeats shutdown drain), N6 (no live-delta coalescing)
Low	5	N4 (validator gaps), N5 (resync/op race), N7 (aggregator lifecycle nits), N8 (standby aggregator + channel leak on site delete), N9 (dual-endpoint requirement)

Both the Critical and the High are confined to the peer-join anti-entropy resync (PLAN-02 T20/T21) — the one round-2 feature that shipped without a two-node integration test — and both are small, well-localized fixes: swap two predicates to the existing shared helper, and chunk one message.