fix(saf): standby assembles chunked resync, applies atomically with re-check, acks; N5 race documented (plan R2-02 T6)

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.
 /// </summary>
-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;
 
+    /// <summary>Akka timer scheduler injected by the framework via <see cref="IWithTimers"/>.</summary>
+    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<int, List<StoreAndForwardMessage>> _assemblingChunks = new();
+    private const string ResyncAssemblyTimerKey = "sf-resync-assembly-timeout";
+
+    /// <summary>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.</summary>
+    private readonly TimeSpan _resyncAssemblyTimeout;
+
     /// <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
@@ -113,7 +128,8 @@ public class SiteReplicationActor : ReceiveActor
         IDeploymentConfigFetcher? configFetcher = null,
         Func<bool>? 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<RequestSfBufferResync>(HandleRequestSfBufferResync);
         Receive<SfResyncSnapshotLoaded>(HandleSfResyncSnapshotLoaded);
+        Receive<SfBufferSnapshotChunk>(HandleSfBufferSnapshotChunk);
+        Receive<ResyncAssemblyTimedOut>(HandleResyncAssemblyTimedOut);
         Receive<SfBufferSnapshot>(HandleSfBufferSnapshot); // legacy monolithic handler — retained for rolling compat
     }
 
@@ -520,9 +539,97 @@ public class SiteReplicationActor : ReceiveActor
             });
     }
 
+    /// <summary>
+    /// Standby-node side of the chunked anti-entropy resync: accumulates the chunks of one
+    /// <c>ResyncId</c>, and once all have arrived, assembles them in sequence order and
+    /// replaces the local buffer atomically, then acks. A new <c>ResyncId</c> discards any
+    /// stale partial assembly (review 02 round 2, N2). An active node ignores chunks.
+    /// </summary>
+    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();
+    }
+
     /// <summary>Internal: the resync snapshot finished loading; chunk and send to the requester.</summary>
     internal sealed record SfResyncSnapshotLoaded(
         IActorRef ReplyTo, List<StoreAndForwardMessage> Messages, bool Truncated);
+
+    /// <summary>Internal: a partial chunk assembly for <paramref name="ResyncId"/> exceeded its window.</summary>
+    internal sealed record ResyncAssemblyTimedOut(string ResyncId);
 }
 
 /// <summary>

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<StoreAndForwardMessage> { NewMessage("f1") }, false), TestActor);
+        actor.Tell(new SfBufferSnapshotChunk(resyncId, 2, 2,
+            new List<StoreAndForwardMessage> { NewMessage("f2") }, false), TestActor);
+
+        var ack = ExpectMsg<SfBufferResyncAck>(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<StoreAndForwardMessage> { NewMessage("orphan") }, false), TestActor);
+        actor.Tell(new SfBufferSnapshotChunk("new", 1, 1,
+            new List<StoreAndForwardMessage> { NewMessage("fresh") }, false), TestActor);
+
+        ExpectMsg<SfBufferResyncAck>(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<StoreAndForwardMessage> { NewMessage("x") }, false), TestActor);
+        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+    }
+
     private static StoreAndForwardMessage NewSfMessage(string id) => new()
     {
         Id = id,