diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs index de61abef..49930016 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs @@ -18,7 +18,7 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; /// Inbound: receives replicated operations from peer and applies to local SQLite. /// Uses fire-and-forget (Tell) — no ack wait per design. /// -public class SiteReplicationActor : ReceiveActor +public class SiteReplicationActor : ReceiveActor, IWithTimers { private readonly SiteStorageService _storage; private readonly StoreAndForwardStorage _sfStorage; @@ -32,6 +32,21 @@ public class SiteReplicationActor : ReceiveActor private readonly TimeSpan _configFetchRetryDelay; private Address? _peerAddress; + /// Akka timer scheduler injected by the framework via . + public ITimerScheduler Timers { get; set; } = null!; + + // ── Chunked-resync assembly (standby side; actor-thread only) ── + private string? _assemblingResyncId; + private int _assemblingTotalChunks; + private bool _assemblingTruncated; + private readonly Dictionary> _assemblingChunks = new(); + private const string ResyncAssemblyTimerKey = "sf-resync-assembly-timeout"; + + /// How long a partial chunk assembly may wait for its missing chunks before + /// being discarded (a lost chunk = lost resync; the next peer-track retries). Ctor + /// test seam; production default 30 s. + private readonly TimeSpan _resyncAssemblyTimeout; + /// /// Maximum rows an active node returns in a single anti-entropy resync snapshot. /// A standby whose buffer exceeded this (Truncated snapshot) resyncs the oldest @@ -113,7 +128,8 @@ public class SiteReplicationActor : ReceiveActor IDeploymentConfigFetcher? configFetcher = null, Func? isActiveOverride = null, SiteRuntimeOptions? options = null, - TimeSpan? configFetchRetryDelay = null) + TimeSpan? configFetchRetryDelay = null, + TimeSpan? resyncAssemblyTimeout = null) { _storage = storage; _sfStorage = sfStorage; @@ -123,6 +139,7 @@ public class SiteReplicationActor : ReceiveActor _logger = logger; _cluster = Cluster.Get(Context.System); _isActive = isActiveOverride ?? DefaultIsActive; + _resyncAssemblyTimeout = resyncAssemblyTimeout ?? TimeSpan.FromSeconds(30); // UA2: bound the standby's replicated-config fetch retries. At least one // attempt always runs; the fixed inter-attempt delay is a test seam // (production default 2 s). @@ -154,6 +171,8 @@ public class SiteReplicationActor : ReceiveActor // Anti-entropy — full S&F buffer resync on peer (re)join Receive(HandleRequestSfBufferResync); Receive(HandleSfResyncSnapshotLoaded); + Receive(HandleSfBufferSnapshotChunk); + Receive(HandleResyncAssemblyTimedOut); Receive(HandleSfBufferSnapshot); // legacy monolithic handler — retained for rolling compat } @@ -520,9 +539,97 @@ public class SiteReplicationActor : ReceiveActor }); } + /// + /// Standby-node side of the chunked anti-entropy resync: accumulates the chunks of one + /// ResyncId, and once all have arrived, assembles them in sequence order and + /// replaces the local buffer atomically, then acks. A new ResyncId discards any + /// stale partial assembly (review 02 round 2, N2). An active node ignores chunks. + /// + private void HandleSfBufferSnapshotChunk(SfBufferSnapshotChunk msg) + { + if (SafeIsActive()) + { + _logger.LogDebug("Ignoring S&F resync chunk — this node is active"); + return; + } + + if (_assemblingResyncId != msg.ResyncId) + { + if (_assemblingResyncId != null) + _logger.LogWarning( + "Discarding partial S&F resync assembly {Old} ({Have}/{Want} chunks): a new resync {New} superseded it", + _assemblingResyncId, _assemblingChunks.Count, _assemblingTotalChunks, msg.ResyncId); + _assemblingResyncId = msg.ResyncId; + _assemblingTotalChunks = msg.TotalChunks; + _assemblingTruncated = msg.Truncated; + _assemblingChunks.Clear(); + } + + _assemblingChunks[msg.Sequence] = msg.Messages; + Timers.StartSingleTimer(ResyncAssemblyTimerKey, new ResyncAssemblyTimedOut(msg.ResyncId), _resyncAssemblyTimeout); + + if (_assemblingChunks.Count < _assemblingTotalChunks) + return; + + // Complete: assemble in sequence order and apply atomically. + var assembled = Enumerable.Range(1, _assemblingTotalChunks) + .SelectMany(seq => _assemblingChunks[seq]) + .ToList(); + var resyncId = _assemblingResyncId!; + var truncated = _assemblingTruncated; + _assemblingResyncId = null; + _assemblingChunks.Clear(); + Timers.Cancel(ResyncAssemblyTimerKey); + + if (truncated) + _logger.LogWarning( + "S&F buffer resync snapshot truncated at {Cap} rows; divergence beyond the cap drains naturally", + MaxResyncRows); + _logger.LogInformation( + "Applying chunked S&F resync {ResyncId} ({Count} rows), replacing local buffer", resyncId, assembled.Count); + + // KNOWN, ACCEPTED race (review 02 round 2, N5 — do NOT "fix" this into something + // worse): a replicated Remove sent after the active node read its snapshot but + // before the snapshot's chunks is ordered BEFORE them on the wire (same + // sender/receiver pair), so this apply can re-add the removed row → an orphan + // Pending row that a later failover re-delivers ONCE. Bounded, self-correcting at + // the next resync, and inherent to no-ack replication; a delivered-side dedup or + // op-sequencing scheme would cost far more than one rare duplicate. + var replyTo = Sender; + Task.Run(async () => + { + if (SafeIsActive()) // belt-and-braces, mirrors the monolithic path (T3) + { + _logger.LogWarning("Discarding chunked S&F resync {ResyncId}: node became active before apply", resyncId); + return; + } + await _sfStorage.ReplaceAllAsync(assembled); + replyTo.Tell(new SfBufferResyncAck(resyncId, assembled.Count)); + }) + .ContinueWith(t => + { + if (t.IsFaulted) + _logger.LogError(t.Exception, "Failed to apply chunked S&F resync {ResyncId}", resyncId); + }); + } + + private void HandleResyncAssemblyTimedOut(ResyncAssemblyTimedOut msg) + { + if (_assemblingResyncId != msg.ResyncId) return; // superseded already + _logger.LogWarning( + "S&F resync assembly {ResyncId} timed out with {Have}/{Want} chunks — discarding partial (a lost chunk; the next peer-track retries)", + msg.ResyncId, _assemblingChunks.Count, _assemblingTotalChunks); + ScadaBridgeTelemetry.RecordReplicationFailure(); + _assemblingResyncId = null; + _assemblingChunks.Clear(); + } + /// Internal: the resync snapshot finished loading; chunk and send to the requester. internal sealed record SfResyncSnapshotLoaded( IActorRef ReplyTo, List Messages, bool Truncated); + + /// Internal: a partial chunk assembly for exceeded its window. + internal sealed record ResyncAssemblyTimedOut(string ResyncId); } /// 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 2ccd9ea4..32a0a7b1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs @@ -335,6 +335,57 @@ akka { Assert.Equal(3, rest.Sum(c => c.Messages.Count)); } + // ── R2 T6: standby chunk assembly + atomic apply + ack ── + + [Fact] + public async Task StandbyNode_AssemblesChunks_AppliesOnce_AndAcks() + { + await _sfStorage.EnqueueAsync(NewMessage("stale")); + var actor = CreateResyncActor(isActive: () => false); + var resyncId = "r1"; + + actor.Tell(new SfBufferSnapshotChunk(resyncId, 1, 2, + new List { NewMessage("f1") }, false), TestActor); + actor.Tell(new SfBufferSnapshotChunk(resyncId, 2, 2, + new List { NewMessage("f2") }, false), TestActor); + + var ack = ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal(resyncId, ack.ResyncId); + Assert.Equal(2, ack.RowCount); + await AwaitAssertAsync(async () => + { + Assert.Null(await _sfStorage.GetMessageByIdAsync("stale")); // replaced wholesale + Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f1")); + Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f2")); + }); + } + + [Fact] + public async Task StandbyNode_NewResyncId_DiscardsStalePartialAssembly() + { + var actor = CreateResyncActor(isActive: () => false); + actor.Tell(new SfBufferSnapshotChunk("old", 1, 2, + new List { NewMessage("orphan") }, false), TestActor); + actor.Tell(new SfBufferSnapshotChunk("new", 1, 1, + new List { NewMessage("fresh") }, false), TestActor); + + ExpectMsg(TimeSpan.FromSeconds(5)); // "new" completed + await AwaitAssertAsync(async () => + { + Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh")); + Assert.Null(await _sfStorage.GetMessageByIdAsync("orphan")); // stale partial never applied + }); + } + + [Fact] + public void ActiveNode_IgnoresChunks_NeverAcks() + { + var actor = CreateResyncActor(isActive: () => true); + actor.Tell(new SfBufferSnapshotChunk("r", 1, 1, + new List { NewMessage("x") }, false), TestActor); + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + private static StoreAndForwardMessage NewSfMessage(string id) => new() { Id = id,