From 3a5b885a447c3b77637c8b29fd5a534223d23328 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 09:48:37 -0400 Subject: [PATCH] fix(saf): resync authority uses the shared oldest-Up predicate + delivery-gate delegate; apply-time re-check guard (plan R2-02 T3) --- .../Actors/AkkaHostedService.cs | 21 +++++++-- .../Actors/SiteReplicationActor.cs | 47 +++++++++++-------- 2 files changed, 44 insertions(+), 24 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index 7f060921..b28c21ba 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -772,10 +772,20 @@ akka {{ var replicationLogger = _serviceProvider.GetRequiredService() .CreateLogger(); + // ONE active-node predicate instance governs the S&F delivery gate, the resync + // authority checks (SiteReplicationActor), and the heartbeat IsActive stamp + // (SiteCommunicationActor, wired below) — review 02 round 2, N1. Null in + // non-clustered test hosts: the actors fall back to the shared oldest-Up + // evaluator, never to a leader check. + var clusterNodeProvider = _serviceProvider.GetService(); + Func? activeNodeCheck = clusterNodeProvider != null + ? () => clusterNodeProvider.SelfIsPrimary + : null; + var replicationActor = _actorSystem!.ActorOf( Props.Create(() => new SiteReplicationActor( storage, sfStorage, replicationService, siteRole, replicationLogger, - deploymentConfigFetcher, null, siteRuntimeOptionsValue, null)), + deploymentConfigFetcher, activeNodeCheck, siteRuntimeOptionsValue, null)), "site-replication"); // Wire S&F replication handler to forward operations via the replication actor @@ -872,10 +882,11 @@ akka {{ // tick so failover resumes delivery within one RetryTimerInterval. // IClusterNodeProvider.SelfIsPrimary is the canonical "this node is the // oldest Up member (singleton host)" check from the cluster-infrastructure - // fix plan — the shared helper this seam was designed to accept. In a - // non-clustered test host the provider is unregistered, so the gate stays - // unset and the sweep is ungated (legacy behaviour, preserved). - var clusterNodeProvider = _serviceProvider.GetService(); + // fix plan — the shared helper this seam was designed to accept. The provider + // is resolved once above (the same instance gating the resync authority and + // heartbeat IsActive stamp — N1). In a non-clustered test host the provider is + // unregistered, so the gate stays unset and the sweep is ungated (legacy + // behaviour, preserved). if (clusterNodeProvider != null) { storeAndForwardService.SetDeliveryGate(() => clusterNodeProvider.SelfIsPrimary); diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs index 0123b049..6e344bac 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs @@ -54,10 +54,11 @@ public class SiteReplicationActor : ReceiveActor /// config never crosses the intra-site Akka hop. Null on nodes/tests without a fetcher. /// /// - /// Test seam for the active-node check that gates the buffer-resync roles (a - /// standby requests a resync, the active node answers). Null (production) uses - /// the repo-standard leader+Up check via — swap point for - /// plan 01's shared active-node helper. + /// Active-node check that gates the buffer-resync roles (a standby requests a + /// resync, the active node answers). Production wiring passes the Host's + /// IClusterNodeProvider.SelfIsPrimary delegate (the same instance gating the + /// S&F delivery sweep); null falls back to the shared oldest-Up evaluator + /// (). /// /// Site runtime options, including the config-fetch retry count; production defaults apply when null. /// Delay between config-fetch retry attempts; defaults to 2 seconds when null. @@ -181,21 +182,17 @@ public class SiteReplicationActor : ReceiveActor } /// - /// Repo-standard active-node check: this node is active when it is the current - /// cluster leader AND its own is - /// . Mirrors SiteCommunicationActor.DefaultIsActiveCheck - /// (swap point for plan 01's shared helper). Any other state reports standby — - /// safe-by-default. + /// Repo-standard active-node check: this node is active when it is the OLDEST Up + /// member carrying the site role — the same oldest-Up semantics as the S&F delivery + /// gate (IClusterNodeProvider.SelfIsPrimary → ClusterActivityEvaluator → shared + /// ActiveNodeEvaluator). NEVER the cluster leader: leadership is lowest-address and + /// diverges from singleton/delivery placement permanently after the lower-address + /// node restarts — the divergence that made the delivering node wipe its own live + /// buffer via a wrong-direction resync (review 02 round 2, N1 Critical). Any other + /// state reports standby — safe-by-default. /// - private bool DefaultIsActive() - { - var self = _cluster.SelfMember; - if (self.Status != MemberStatus.Up) - return false; - - var leader = _cluster.State.Leader; - return leader != null && leader == self.Address; - } + private bool DefaultIsActive() => + Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(_cluster, _siteRole); /// /// Evaluates the active-node check, treating a throwing check as standby @@ -435,7 +432,19 @@ public class SiteReplicationActor : ReceiveActor _logger.LogInformation( "Applying S&F buffer resync snapshot ({Count} rows), replacing local buffer", msg.Messages.Count); - _sfStorage.ReplaceAllAsync(msg.Messages) + Task.Run(async () => + { + // Belt-and-braces (N1): re-check at apply time. ReplaceAllAsync discards + // every in-flight row (StoreAndForwardStorage.cs "Never call on an active + // node"); a flip between message receipt and this point must abort. + if (SafeIsActive()) + { + _logger.LogWarning( + "Discarding S&F buffer resync snapshot: this node became active before apply"); + return; + } + await _sfStorage.ReplaceAllAsync(msg.Messages); + }) .ContinueWith(t => { if (t.IsFaulted)