feat(store-and-forward): peer-join full-buffer resync — standby anti-entropy for the S&F buffer

src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs

@@ -27,8 +27,16 @@ public class SiteReplicationActor : ReceiveActor
     private readonly string _siteRole;
     private readonly ILogger<SiteReplicationActor> _logger;
     private readonly Cluster _cluster;
+    private readonly Func<bool> _isActive;
     private Address? _peerAddress;
 
+    /// <summary>
+    /// Maximum rows an active node returns in a single anti-entropy resync snapshot.
+    /// A standby whose buffer exceeded this (Truncated snapshot) resyncs the oldest
+    /// 10 000 rows; further divergence drains naturally as the active node delivers.
+    /// </summary>
+    private const int MaxResyncRows = 10_000;
+
     /// <summary>
     /// Initializes a new <see cref="SiteReplicationActor"/> and registers Akka message handlers.
     /// </summary>
@@ -43,13 +51,20 @@ public class SiteReplicationActor : ReceiveActor
     /// replicates only the deployment id, and the standby fetches the config itself so a large
     /// config never crosses the intra-site Akka hop. Null on nodes/tests without a fetcher.
     /// </param>
+    /// <param name="isActiveOverride">
+    /// 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 <see cref="Cluster"/> — swap point for
+    /// plan 01's shared active-node helper.
+    /// </param>
     public SiteReplicationActor(
         SiteStorageService storage,
         StoreAndForwardStorage sfStorage,
         ReplicationService replicationService,
         string siteRole,
         ILogger<SiteReplicationActor> logger,
-        IDeploymentConfigFetcher? configFetcher = null)
+        IDeploymentConfigFetcher? configFetcher = null,
+        Func<bool>? isActiveOverride = null)
     {
         _storage = storage;
         _sfStorage = sfStorage;
@@ -58,6 +73,7 @@ public class SiteReplicationActor : ReceiveActor
         _siteRole = siteRole;
         _logger = logger;
         _cluster = Cluster.Get(Context.System);
+        _isActive = isActiveOverride ?? DefaultIsActive;
 
         // Cluster member events
         Receive<ClusterEvent.MemberUp>(HandleMemberUp);
@@ -80,6 +96,10 @@ public class SiteReplicationActor : ReceiveActor
         Receive<ApplyConfigSetEnabled>(HandleApplyConfigSetEnabled);
         Receive<ApplyArtifacts>(HandleApplyArtifacts);
         Receive<ApplyStoreAndForward>(HandleApplyStoreAndForward);
+
+        // Anti-entropy — full S&F buffer resync on peer (re)join
+        Receive<RequestSfBufferResync>(HandleRequestSfBufferResync);
+        Receive<SfBufferSnapshot>(HandleSfBufferSnapshot);
     }
 
     /// <inheritdoc />
@@ -129,6 +149,57 @@ public class SiteReplicationActor : ReceiveActor
         {
             _peerAddress = member.Address;
             _logger.LogInformation("Peer node tracked: {Address}", _peerAddress);
+            OnPeerTracked();
+        }
+    }
+
+    /// <summary>
+    /// Side-effect run whenever a peer is (re)tracked. A <b>standby</b> requests a
+    /// full S&amp;F buffer snapshot for anti-entropy resync — this closes the "a
+    /// standby down for an hour rejoins and diverges forever" gap: it may have missed
+    /// replicated Add/Remove/Park ops while it was gone. The active node never
+    /// requests. <see langword="protected virtual"/> so tests can drive it without a
+    /// real two-node cluster.
+    /// </summary>
+    protected virtual void OnPeerTracked()
+    {
+        if (!SafeIsActive())
+        {
+            SendToPeer(new RequestSfBufferResync());
+        }
+    }
+
+    /// <summary>
+    /// Repo-standard active-node check: this node is active when it is the current
+    /// cluster leader AND its own <see cref="MemberStatus"/> is
+    /// <see cref="MemberStatus.Up"/>. Mirrors <c>SiteCommunicationActor.DefaultIsActiveCheck</c>
+    /// (swap point for plan 01's shared helper). Any other state reports standby —
+    /// safe-by-default.
+    /// </summary>
+    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;
+    }
+
+    /// <summary>
+    /// Evaluates the active-node check, treating a throwing check as standby
+    /// (safe-by-default: a standby never delivers or answers resyncs).
+    /// </summary>
+    private bool SafeIsActive()
+    {
+        try
+        {
+            return _isActive();
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Active-node check threw; treating node as standby");
+            return false;
         }
     }
 
@@ -295,4 +366,73 @@ public class SiteReplicationActor : ReceiveActor
                     _logger.LogError(t.Exception, "Failed to apply replicated S&F operation {Id}", msg.Operation.MessageId);
             });
     }
+
+    /// <summary>
+    /// Active-node side of the anti-entropy resync: answers a standby's
+    /// <see cref="RequestSfBufferResync"/> with a full-buffer <see cref="SfBufferSnapshot"/>
+    /// (up to <see cref="MaxResyncRows"/> oldest rows). A non-active node ignores the
+    /// request — only the authoritative node may answer.
+    /// </summary>
+    private void HandleRequestSfBufferResync(RequestSfBufferResync msg)
+    {
+        if (!SafeIsActive())
+        {
+            _logger.LogDebug("Ignoring S&F buffer resync request — this node is not active");
+            return;
+        }
+
+        var replyTo = Sender;
+        _sfStorage.GetAllMessagesAsync(MaxResyncRows).PipeTo(
+            replyTo,
+            Self,
+            success: result => new SfBufferSnapshot(result.Messages, result.Truncated));
+    }
+
+    /// <summary>
+    /// Standby-node side of the anti-entropy resync: replaces the local buffer
+    /// wholesale with the active node's snapshot. Combined with the upsert-based
+    /// replicated applies (arch review 02, Task 11), any replicated op that lands
+    /// after this resync merges cleanly onto the resynced state. An active node
+    /// ignores a snapshot — it is the source of truth, never a resync target.
+    /// </summary>
+    private void HandleSfBufferSnapshot(SfBufferSnapshot msg)
+    {
+        if (SafeIsActive())
+        {
+            _logger.LogDebug("Ignoring S&F buffer snapshot — this node is active");
+            return;
+        }
+
+        if (msg.Truncated)
+        {
+            _logger.LogWarning(
+                "S&F buffer resync snapshot truncated at {Cap} rows; divergence beyond the cap drains naturally",
+                MaxResyncRows);
+        }
+
+        _logger.LogInformation(
+            "Applying S&F buffer resync snapshot ({Count} rows), replacing local buffer", msg.Messages.Count);
+
+        _sfStorage.ReplaceAllAsync(msg.Messages)
+            .ContinueWith(t =>
+            {
+                if (t.IsFaulted)
+                    _logger.LogError(t.Exception, "Failed to apply S&F buffer resync snapshot");
+            });
+    }
 }
+
+/// <summary>
+/// Standby→active: request a full S&amp;F buffer snapshot for anti-entropy resync
+/// (sent when a standby (re)tracks a peer). Crosses Akka remoting between the two
+/// site nodes; the POCO rides the default serializer.
+/// </summary>
+public sealed record RequestSfBufferResync;
+
+/// <summary>
+/// Active→standby: full-buffer snapshot. <paramref name="Truncated"/> is true when
+/// the active node's buffer exceeded <c>MaxResyncRows</c> (the standby logs a Warning —
+/// divergence beyond the cap drains naturally as the active node delivers). Crosses
+/// Akka remoting; the message list rides the default serializer.
+/// </summary>
+public sealed record SfBufferSnapshot(List<StoreAndForwardMessage> Messages, bool Truncated);