From 26dce8b69f6820f254f5e2cedc5af0dc3447b095 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 09:50:44 -0400 Subject: [PATCH] fix(comm): heartbeat IsActive uses the shared oldest-Up predicate; Host wires one delegate everywhere (plan R2-02 T4) --- .../requirements/Component-StoreAndForward.md | 2 +- .../Actors/SiteCommunicationActor.cs | 34 +++++++------------ .../Actors/AkkaHostedService.cs | 3 +- 3 files changed, 16 insertions(+), 23 deletions(-) diff --git a/docs/requirements/Component-StoreAndForward.md b/docs/requirements/Component-StoreAndForward.md index a2d39f56..53aa0564 100644 --- a/docs/requirements/Component-StoreAndForward.md +++ b/docs/requirements/Component-StoreAndForward.md @@ -80,7 +80,7 @@ There is **no maximum buffer size**. Messages accumulate in the buffer until del - The standby node applies the same operations to its own local SQLite database but is **passive**: it never runs the delivery sweep. The retry sweep is **gated to the active node** (the oldest Up member / singleton host, re-evaluated every sweep tick), so only one node delivers at a time. The standby applies replicated operations purely to keep its copy warm for a future failover. - On failover, the new active node has a near-complete copy of the buffer. In rare cases, the most recent operations may not have been replicated (e.g., a message added or removed just before failover). This can result in a few **duplicate deliveries** (message delivered but its `Remove` not yet replicated) or a few **missed retries** (message added but not replicated). Duplicate deliveries are therefore confined to the **failover window** — an in-flight delivery whose `Remove` had not yet replicated — and never occur in steady-state operation (the standby's gate keeps it from delivering the same rows). Both are acceptable trade-offs for the latency benefit. - On failover, the new active node's gate flips to active within one sweep interval and it resumes delivery from its local copy. -- **Peer-join anti-entropy resync.** Asynchronous, no-ack replication keeps the standby warm in steady state, but a standby that was **down for an extended period** (a crash, a long maintenance window) misses every operation replicated while it was gone and would otherwise diverge from the active node's buffer forever. To close that gap, whenever a node **(re)tracks its peer**, a **standby** requests a full-buffer snapshot (`RequestSfBufferResync`); the **active** node answers with up to `MaxResyncRows` (10 000) of its oldest rows (`SfBufferSnapshot`), and the standby **replaces its entire local buffer** with that snapshot (`ReplaceAllAsync`, one transaction). Only the active node answers; only a standby applies (each side checks the repo-standard leader+Up active-node predicate, safe-by-default to standby). Because replicated applies are **upserts** (see the replication apply path), any Add/Remove/Park that lands after the resync merges cleanly onto the resynced state — no primary-key conflict, no lost delta. If the buffer exceeds the 10 000-row cap the snapshot is flagged `Truncated` and the standby logs a Warning; the residual divergence beyond the cap drains naturally as the active node delivers. +- **Peer-join anti-entropy resync.** Asynchronous, no-ack replication keeps the standby warm in steady state, but a standby that was **down for an extended period** (a crash, a long maintenance window) misses every operation replicated while it was gone and would otherwise diverge from the active node's buffer forever. To close that gap, whenever a node **(re)tracks its peer**, a **standby** requests a full-buffer snapshot (`RequestSfBufferResync`); the **active** node answers with up to `MaxResyncRows` (10 000) of its oldest rows (`SfBufferSnapshot`), and the standby **replaces its entire local buffer** with that snapshot (`ReplaceAllAsync`, one transaction). Only the active node answers; only a standby applies (each side checks the repo-standard oldest-Up member active-node predicate — singleton-host semantics via the shared `ActiveNodeEvaluator`, the same predicate as the S&F delivery gate; N1 — safe-by-default to standby). Because replicated applies are **upserts** (see the replication apply path), any Add/Remove/Park that lands after the resync merges cleanly onto the resynced state — no primary-key conflict, no lost delta. If the buffer exceeds the 10 000-row cap the snapshot is flagged `Truncated` and the standby logs a Warning; the residual divergence beyond the cap drains naturally as the active node delivers. ### Operation Tracking Table (lives in Site Runtime, not here) diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs index b3027ac6..4667b2ec 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs @@ -67,10 +67,10 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers /// Local reference to the Deployment Manager singleton proxy. /// /// Optional override returning true when this node - /// is the active member of the site cluster. null uses the real - /// Akka leader check (the default for production - /// wiring); tests pass a stub so they do not need to load Akka.Cluster - /// into the TestKit ActorSystem. + /// is the active member of the site cluster. null uses the shared oldest-Up + /// evaluator (production wiring passes the Host's singleton-host delegate); tests + /// pass a stub so they do not need to load Akka.Cluster into the TestKit + /// ActorSystem. /// public SiteCommunicationActor( string siteId, @@ -506,24 +506,16 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers } /// - /// Default active-node check used when no override is - /// supplied. Mirrors ActiveNodeGate in the Host (and - /// ActiveNodeHealthCheck): the node is the active member of the - /// site cluster when it is the current cluster leader AND its own - /// is . Any other - /// state (still joining, leaving, no leader yet) reports standby — - /// safe-by-default, matching the standby case. + /// Default active-node check used when no override is supplied: oldest-Up member + /// semantics via the shared — the + /// same predicate as the S&F delivery gate and the replication resync authority + /// (review 02 round 2, N1). Unscoped by role: a site cluster's members all carry + /// the site role, so role scoping is a no-op here; production wiring passes the + /// Host's role-scoped IClusterNodeProvider delegate anyway. Any other state + /// (still joining, leaving) reports standby — safe-by-default. /// - private bool DefaultIsActiveCheck() - { - var cluster = Cluster.Get(Context.System); - var self = cluster.SelfMember; - if (self.Status != MemberStatus.Up) - return false; - - var leader = cluster.State.Leader; - return leader != null && leader == self.Address; - } + private bool DefaultIsActiveCheck() => + ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(Cluster.Get(Context.System)); // ── Internal messages ── diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index b28c21ba..7a7d8517 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -833,7 +833,8 @@ akka {{ Props.Create(() => new SiteCommunicationActor( _nodeOptions.SiteId!, _communicationOptions, - dmProxy)), + dmProxy, + activeNodeCheck)), "site-communication"); // Register local handlers with SiteCommunicationActor