diff --git a/docs/requirements/Component-StoreAndForward.md b/docs/requirements/Component-StoreAndForward.md index 2aa69b32..a2d39f56 100644 --- a/docs/requirements/Component-StoreAndForward.md +++ b/docs/requirements/Component-StoreAndForward.md @@ -80,6 +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. ### Operation Tracking Table (lives in Site Runtime, not here) diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs index bc321e0a..af6ee5c4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs @@ -27,8 +27,16 @@ public class SiteReplicationActor : ReceiveActor private readonly string _siteRole; private readonly ILogger _logger; private readonly Cluster _cluster; + private readonly Func _isActive; private Address? _peerAddress; + /// + /// 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. + /// + private const int MaxResyncRows = 10_000; + /// /// Initializes a new and registers Akka message handlers. /// @@ -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. /// + /// + /// 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. + /// public SiteReplicationActor( SiteStorageService storage, StoreAndForwardStorage sfStorage, ReplicationService replicationService, string siteRole, ILogger logger, - IDeploymentConfigFetcher? configFetcher = null) + IDeploymentConfigFetcher? configFetcher = null, + Func? 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(HandleMemberUp); @@ -80,6 +96,10 @@ public class SiteReplicationActor : ReceiveActor Receive(HandleApplyConfigSetEnabled); Receive(HandleApplyArtifacts); Receive(HandleApplyStoreAndForward); + + // Anti-entropy — full S&F buffer resync on peer (re)join + Receive(HandleRequestSfBufferResync); + Receive(HandleSfBufferSnapshot); } /// @@ -129,6 +149,57 @@ public class SiteReplicationActor : ReceiveActor { _peerAddress = member.Address; _logger.LogInformation("Peer node tracked: {Address}", _peerAddress); + OnPeerTracked(); + } + } + + /// + /// Side-effect run whenever a peer is (re)tracked. A standby requests a + /// full S&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. so tests can drive it without a + /// real two-node cluster. + /// + protected virtual void OnPeerTracked() + { + if (!SafeIsActive()) + { + SendToPeer(new RequestSfBufferResync()); + } + } + + /// + /// 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. + /// + 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; + } + + /// + /// Evaluates the active-node check, treating a throwing check as standby + /// (safe-by-default: a standby never delivers or answers resyncs). + /// + 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); }); } + + /// + /// Active-node side of the anti-entropy resync: answers a standby's + /// with a full-buffer + /// (up to oldest rows). A non-active node ignores the + /// request — only the authoritative node may answer. + /// + 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)); + } + + /// + /// 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. + /// + 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"); + }); + } } + +/// +/// Standby→active: request a full S&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. +/// +public sealed record RequestSfBufferResync; + +/// +/// Active→standby: full-buffer snapshot. is true when +/// the active node's buffer exceeded MaxResyncRows (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. +/// +public sealed record SfBufferSnapshot(List Messages, bool Truncated); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs index d9662d49..2fb8910b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs @@ -8,6 +8,7 @@ using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; using ZB.MOM.WW.ScadaBridge.StoreAndForward; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors; @@ -184,6 +185,107 @@ akka { Assert.Equal("tok-abc", applied.FetchToken); } + // ── Task 21: peer-join S&F buffer resync (anti-entropy) ── + + [Fact] + public void StandbyTrackingPeer_SendsResyncRequest() + { + var probe = CreateTestProbe(); + var actor = ActorOf(Props.Create(() => new ResyncTestActor( + _storage, _sfStorage, _replicationService, SiteRole, + NullLogger.Instance, probe.Ref, () => false))); + + actor.Tell(new TriggerPeerTracked()); // stands in for TryTrackPeer's MemberUp path + + probe.ExpectMsg(TimeSpan.FromSeconds(3)); + } + + [Fact] + public void ActiveTrackingPeer_DoesNotRequestResync() + { + var probe = CreateTestProbe(); + var actor = ActorOf(Props.Create(() => new ResyncTestActor( + _storage, _sfStorage, _replicationService, SiteRole, + NullLogger.Instance, probe.Ref, () => true))); + + actor.Tell(new TriggerPeerTracked()); + + probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // active node never requests a resync + } + + [Fact] + public async Task ActiveNode_AnswersResyncRequest_WithFullBufferSnapshot() + { + await _sfStorage.EnqueueAsync(NewSfMessage("m1")); + var probe = CreateTestProbe(); + var actor = ActorOf(Props.Create(() => new ResyncTestActor( + _storage, _sfStorage, _replicationService, SiteRole, + NullLogger.Instance, probe.Ref, () => true))); + + actor.Tell(new RequestSfBufferResync(), TestActor); + + var snapshot = ExpectMsg(TimeSpan.FromSeconds(3)); + Assert.Single(snapshot.Messages); + Assert.False(snapshot.Truncated); + } + + [Fact] + public async Task StandbyNode_AppliesSnapshot_ReplacingItsBuffer() + { + await _sfStorage.EnqueueAsync(NewSfMessage("stale")); + var probe = CreateTestProbe(); + var actor = ActorOf(Props.Create(() => new ResyncTestActor( + _storage, _sfStorage, _replicationService, SiteRole, + NullLogger.Instance, probe.Ref, () => false))); + + actor.Tell(new SfBufferSnapshot(new List { NewSfMessage("fresh") }, false)); + + await AwaitAssertAsync(async () => + { + Assert.Null(await _sfStorage.GetMessageByIdAsync("stale")); + Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh")); + }, TimeSpan.FromSeconds(5)); + } + + private static StoreAndForwardMessage NewSfMessage(string id) => new() + { + Id = id, + Category = StoreAndForwardCategory.ExternalSystem, + Target = "t", + PayloadJson = "{}", + RetryCount = 0, + MaxRetries = 50, + RetryIntervalMs = 30000, + CreatedAt = DateTimeOffset.UtcNow, + Status = StoreAndForwardMessageStatus.Pending, + }; + + /// Test message: drives directly, + /// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer). + private sealed record TriggerPeerTracked; + + /// + /// Test subclass for the resync tests: captures peer sends to a probe, injects the + /// active-node check, and exposes via a test message. + /// + private sealed class ResyncTestActor : SiteReplicationActor + { + private readonly IActorRef _peerProbe; + + public ResyncTestActor( + SiteStorageService storage, StoreAndForwardStorage sfStorage, + ReplicationService replicationService, string siteRole, + ILogger logger, IActorRef peerProbe, Func isActive) + : base(storage, sfStorage, replicationService, siteRole, logger, + configFetcher: null, isActiveOverride: isActive) + { + _peerProbe = peerProbe; + Receive(_ => OnPeerTracked()); + } + + protected override void SendToPeer(object message) => _peerProbe.Tell(message, Self); + } + /// /// Test subclass exposing the peer send: is /// overridden to forward to a probe so the outbound mapping can be asserted without a real