fix(saf): chunk the resync snapshot to fit Akka remoting frames (additive protocol, byte-budgeted) (plan R2-02 T5)

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

@@ -39,6 +39,48 @@ public class SiteReplicationActor : ReceiveActor
     /// </summary>
     private const int MaxResyncRows = 10_000;
 
+    /// <summary>
+    /// Estimated per-chunk payload budget for a resync snapshot. Akka remoting's default
+    /// <c>maximum-frame-size</c> is 128 000 bytes and <c>BuildHocon</c> sets no override,
+    /// so the monolithic <see cref="SfBufferSnapshot"/> is silently undeliverable for any
+    /// realistic backlog (review 02 round 2, N2). 64 000 bytes leaves ≈50% headroom for
+    /// the JSON envelope, CLR type manifests, and the non-payload columns.
+    /// </summary>
+    internal const int MaxResyncChunkBytes = 64_000;
+
+    /// <summary>Row cap per resync chunk (bounds a chunk even when every row is tiny).</summary>
+    internal const int MaxResyncChunkRows = 200;
+
+    /// <summary>
+    /// Splits a resync snapshot into chunks that fit Akka remoting's default
+    /// 128 000-byte frame (review 02 round 2, N2): rows accumulate until the estimated
+    /// payload budget or the row cap is hit. Estimation is payload-dominated
+    /// (payload_json length + 512 bytes fixed overhead per row); a single row whose
+    /// payload exceeds the budget ships alone (Warning at the call site). Order is
+    /// preserved (oldest-first, matching GetAllMessagesAsync).
+    /// </summary>
+    internal static List<List<StoreAndForwardMessage>> ChunkForRemoting(
+        IReadOnlyList<StoreAndForwardMessage> rows, int maxChunkBytes, int maxChunkRows)
+    {
+        var chunks = new List<List<StoreAndForwardMessage>>();
+        var current = new List<StoreAndForwardMessage>();
+        var currentBytes = 0;
+        foreach (var row in rows)
+        {
+            var estimate = (row.PayloadJson?.Length ?? 0) + 512;
+            if (current.Count > 0 && (currentBytes + estimate > maxChunkBytes || current.Count >= maxChunkRows))
+            {
+                chunks.Add(current);
+                current = new List<StoreAndForwardMessage>();
+                currentBytes = 0;
+            }
+            current.Add(row);
+            currentBytes += estimate;
+        }
+        if (current.Count > 0) chunks.Add(current);
+        return chunks;
+    }
+
     /// <summary>
     /// Initializes a new <see cref="SiteReplicationActor"/> and registers Akka message handlers.
     /// </summary>
@@ -111,7 +153,8 @@ public class SiteReplicationActor : ReceiveActor
 
         // Anti-entropy — full S&F buffer resync on peer (re)join
         Receive<RequestSfBufferResync>(HandleRequestSfBufferResync);
-        Receive<SfBufferSnapshot>(HandleSfBufferSnapshot);
+        Receive<SfResyncSnapshotLoaded>(HandleSfResyncSnapshotLoaded);
+        Receive<SfBufferSnapshot>(HandleSfBufferSnapshot); // legacy monolithic handler — retained for rolling compat
     }
 
     /// <inheritdoc />
@@ -388,9 +431,10 @@ public class SiteReplicationActor : ReceiveActor
 
     /// <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.
+    /// <see cref="RequestSfBufferResync"/> with a sequence of byte-budgeted
+    /// <see cref="SfBufferSnapshotChunk"/>s (up to <see cref="MaxResyncRows"/> oldest rows).
+    /// A non-active node ignores the request — only the authoritative node may answer.
+    /// The snapshot is piped back to Self so chunking + ack bookkeeping stays actor-safe.
     /// </summary>
     private void HandleRequestSfBufferResync(RequestSfBufferResync msg)
     {
@@ -402,9 +446,33 @@ public class SiteReplicationActor : ReceiveActor
 
         var replyTo = Sender;
         _sfStorage.GetAllMessagesAsync(MaxResyncRows).PipeTo(
-            replyTo,
             Self,
-            success: result => new SfBufferSnapshot(result.Messages, result.Truncated));
+            failure: ex => new Status.Failure(ex),
+            success: result => new SfResyncSnapshotLoaded(replyTo, result.Messages, result.Truncated));
+    }
+
+    /// <summary>
+    /// Active-node continuation: the resync snapshot finished loading; chunk it to fit the
+    /// remoting frame and send the sequenced chunks to the requester (all sharing one
+    /// resyncId). Task 7 arms the ack-timeout here.
+    /// </summary>
+    private void HandleSfResyncSnapshotLoaded(SfResyncSnapshotLoaded msg)
+    {
+        var chunks = ChunkForRemoting(msg.Messages, MaxResyncChunkBytes, MaxResyncChunkRows);
+        if (chunks.Count == 0) chunks.Add(new List<StoreAndForwardMessage>()); // empty buffer still resyncs (clears the standby)
+        var resyncId = Guid.NewGuid().ToString("N");
+        for (var i = 0; i < chunks.Count; i++)
+        {
+            if (chunks[i].Count == 1 && (chunks[i][0].PayloadJson?.Length ?? 0) + 512 > MaxResyncChunkBytes)
+                _logger.LogWarning(
+                    "Resync row {Id} alone exceeds the chunk budget ({Bytes}B payload); sending solo — it may exceed the remoting frame",
+                    chunks[i][0].Id, chunks[i][0].PayloadJson?.Length ?? 0);
+            msg.ReplyTo.Tell(new SfBufferSnapshotChunk(resyncId, i + 1, chunks.Count, chunks[i], msg.Truncated), Self);
+        }
+        _logger.LogInformation(
+            "Answered S&F resync request with {Rows} row(s) in {Chunks} chunk(s), resyncId={ResyncId}",
+            msg.Messages.Count, chunks.Count, resyncId);
+        // Task 7 registers the pending-ack entry here.
     }
 
     /// <summary>
@@ -451,6 +519,10 @@ public class SiteReplicationActor : ReceiveActor
                     _logger.LogError(t.Exception, "Failed to apply S&F buffer resync snapshot");
             });
     }
+
+    /// <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>
@@ -467,3 +539,24 @@ public sealed record RequestSfBufferResync;
 /// Akka remoting; the message list rides the default serializer.
 /// </summary>
 public sealed record SfBufferSnapshot(List<StoreAndForwardMessage> Messages, bool Truncated);
+
+/// <summary>
+/// Active→standby: one sequenced chunk of a full-buffer anti-entropy snapshot
+/// (review 02 round 2, N2 — the monolithic <see cref="SfBufferSnapshot"/> exceeds Akka
+/// remoting's default 128 000-byte frame for any realistic backlog). All chunks of one
+/// resync share <paramref name="ResyncId"/>; <paramref name="Sequence"/> is 1-based up to
+/// <paramref name="TotalChunks"/>. Additive message — the legacy monolithic snapshot
+/// handler is retained for rolling upgrades. Crosses intra-site Akka remoting (NOT
+/// ClusterClient — ClusterClientContractLockTests is intentionally not involved).
+/// </summary>
+public sealed record SfBufferSnapshotChunk(
+    string ResyncId, int Sequence, int TotalChunks,
+    List<StoreAndForwardMessage> Messages, bool Truncated);
+
+/// <summary>
+/// Standby→active: delivery confirmation — the standby assembled all chunks of
+/// <paramref name="ResyncId"/> and applied them atomically (<paramref name="RowCount"/>
+/// rows installed). Absence within the ack window is surfaced by the active node
+/// (Warning + counter) — the silent-loss mode N2 flagged.
+/// </summary>
+public sealed record SfBufferResyncAck(string ResyncId, int RowCount);

tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs

@@ -248,8 +248,10 @@ akka {
     }
 
     [Fact]
-    public async Task ActiveNode_AnswersResyncRequest_WithFullBufferSnapshot()
+    public async Task ActiveNode_AnswersResyncRequest_WithChunkedSnapshot()
     {
+        // Post-R2-T5 the active node answers with byte-budgeted SfBufferSnapshotChunk(s)
+        // (a single small row rides one chunk) rather than the monolithic SfBufferSnapshot.
         await _sfStorage.EnqueueAsync(NewSfMessage("m1"));
         var probe = CreateTestProbe();
         var actor = ActorOf(Props.Create(() => new ResyncTestActor(
@@ -258,9 +260,11 @@ akka {
 
         actor.Tell(new RequestSfBufferResync(), TestActor);
 
-        var snapshot = ExpectMsg<SfBufferSnapshot>(TimeSpan.FromSeconds(3));
-        Assert.Single(snapshot.Messages);
-        Assert.False(snapshot.Truncated);
+        var chunk = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(3));
+        Assert.Equal(1, chunk.TotalChunks);
+        Assert.Equal(1, chunk.Sequence);
+        Assert.Single(chunk.Messages);
+        Assert.False(chunk.Truncated);
     }
 
     [Fact]
@@ -281,6 +285,56 @@ akka {
         }, TimeSpan.FromSeconds(5));
     }
 
+    // ── R2 T5: chunked resync answer ──
+
+    [Fact]
+    public void ChunkForRemoting_SplitsByByteBudget_PreservingOrderAndSequence()
+    {
+        var rows = Enumerable.Range(0, 10)
+            .Select(i => NewMessage($"m{i}", payloadJson: new string('x', 20_000)))
+            .ToList();
+
+        var chunks = SiteReplicationActor.ChunkForRemoting(rows, maxChunkBytes: 64_000, maxChunkRows: 200);
+
+        Assert.True(chunks.Count > 1); // 10 × 20 KB cannot ride one 64 KB chunk
+        Assert.Equal(rows.Select(r => r.Id), chunks.SelectMany(c => c).Select(r => r.Id)); // order preserved
+        Assert.All(chunks, c => Assert.True(
+            c.Sum(r => r.PayloadJson.Length) <= 64_000 || c.Count == 1)); // budget honored (oversized row isolated)
+    }
+
+    [Fact]
+    public void ChunkForRemoting_RowCapHonored_AndSingleOversizedRowIsolated()
+    {
+        var many = Enumerable.Range(0, 500).Select(i => NewMessage($"s{i}", payloadJson: "{}")).ToList();
+        Assert.All(SiteReplicationActor.ChunkForRemoting(many, 64_000, 200), c => Assert.True(c.Count <= 200));
+
+        var oversized = new List<StoreAndForwardMessage>
+            { NewMessage("big", payloadJson: new string('y', 100_000)), NewMessage("small", payloadJson: "{}") };
+        var chunks = SiteReplicationActor.ChunkForRemoting(oversized, 64_000, 200);
+        Assert.Equal(2, chunks.Count); // the oversized row rides alone
+    }
+
+    [Fact]
+    public async Task ActiveNode_AnswersResyncRequest_WithSequencedChunks_SharingOneResyncId()
+    {
+        for (var i = 0; i < 3; i++)
+            await _sfStorage.EnqueueAsync(NewMessage($"c{i}", payloadJson: new string('z', 30_000)));
+        var actor = CreateResyncActor(isActive: () => true);
+
+        actor.Tell(new RequestSfBufferResync(), TestActor);
+
+        var first = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
+        var rest = Enumerable.Range(1, first.TotalChunks - 1)
+            .Select(_ => ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5)))
+            .Prepend(first)
+            .ToList();
+
+        Assert.True(first.TotalChunks > 1);
+        Assert.All(rest, c => Assert.Equal(first.ResyncId, c.ResyncId));
+        Assert.Equal(Enumerable.Range(1, first.TotalChunks), rest.Select(c => c.Sequence));
+        Assert.Equal(3, rest.Sum(c => c.Messages.Count));
+    }
+
     private static StoreAndForwardMessage NewSfMessage(string id) => new()
     {
         Id = id,
@@ -294,6 +348,30 @@ akka {
         Status = StoreAndForwardMessageStatus.Pending,
     };
 
+    /// <summary>
+    /// Builds a resync-test message with a settable payload (additive to
+    /// <see cref="NewSfMessage"/> — the chunker sizes on <c>PayloadJson</c> length).
+    /// </summary>
+    private static StoreAndForwardMessage NewMessage(string id, string payloadJson = "{}") => new()
+    {
+        Id = id,
+        Category = StoreAndForwardCategory.ExternalSystem,
+        Target = "t",
+        PayloadJson = payloadJson,
+        RetryCount = 0,
+        MaxRetries = 50,
+        RetryIntervalMs = 30000,
+        CreatedAt = DateTimeOffset.UtcNow,
+        Status = StoreAndForwardMessageStatus.Pending,
+    };
+
+    /// <summary>Constructs a <see cref="ResyncTestActor"/> with the given active-node check
+    /// (the resync chunk/ack tests Tell to and expect from <see cref="TestKit.TestActor"/>).</summary>
+    private IActorRef CreateResyncActor(Func<bool> isActive) =>
+        ActorOf(Props.Create(() => new ResyncTestActor(
+            _storage, _sfStorage, _replicationService, SiteRole,
+            NullLogger<SiteReplicationActor>.Instance, CreateTestProbe().Ref, isActive)));
+
     /// <summary>Test message: drives <see cref="SiteReplicationActor.OnPeerTracked"/> directly,
     /// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer).</summary>
     private sealed record TriggerPeerTracked;

tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs

@@ -0,0 +1,95 @@
+using Akka.TestKit.Xunit2;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
+using ZB.MOM.WW.ScadaBridge.StoreAndForward;
+
+namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests;
+
+/// <summary>
+/// Characterization pin for the chunked anti-entropy resync contract (review 02 round 2,
+/// N2). <see cref="SfBufferSnapshotChunk"/> (active→standby) and <see cref="SfBufferResyncAck"/>
+/// (standby→active) ride intra-site Akka remoting on the default reflective-JSON wire
+/// format. A rename/move, dropped setter, or non-default-constructible message would
+/// silently break resync across a rolling upgrade and only surface as a divergent buffer
+/// after a failover. These pin round-trip fidelity and type identity. (These messages are
+/// NOT ClusterClient traffic, so they are intentionally absent from ClusterClientContractLockTests.)
+/// </summary>
+public class ResyncWireSerializationPinTests : TestKit
+{
+    private T RoundTrip<T>(T message)
+    {
+        var serialization = Sys.Serialization;
+        var serializer = serialization.FindSerializerFor(message);
+        var bytes = serializer.ToBinary(message);
+        return (T)serialization.Deserialize(bytes, serializer.Identifier, message!.GetType());
+    }
+
+    private static StoreAndForwardMessage FullMessage() => new()
+    {
+        Id = Guid.NewGuid().ToString("N"),
+        Category = StoreAndForwardCategory.Notification,
+        Target = "Operators",
+        PayloadJson = "{\"notificationId\":\"abc\"}",
+        RetryCount = 4,
+        MaxRetries = 0,
+        RetryIntervalMs = 30000,
+        CreatedAt = DateTimeOffset.UtcNow,
+        LastAttemptAt = DateTimeOffset.UtcNow,
+        Status = StoreAndForwardMessageStatus.Parked,
+        LastError = "central rejected",
+        OriginInstanceName = "Plant.Pump3",
+        ExecutionId = Guid.NewGuid(),
+        SourceScript = "ScriptActor:MonitorSpeed",
+        ParentExecutionId = Guid.NewGuid(),
+    };
+
+    [Fact]
+    public void SfBufferSnapshotChunk_WithFullMessage_RoundTripsOnTheWire()
+    {
+        var message = FullMessage();
+        var original = new SfBufferSnapshotChunk(
+            "resync-1", 2, 5, new List<StoreAndForwardMessage> { message }, Truncated: true);
+
+        var back = RoundTrip(original);
+
+        Assert.Equal(original.ResyncId, back.ResyncId);
+        Assert.Equal(original.Sequence, back.Sequence);
+        Assert.Equal(original.TotalChunks, back.TotalChunks);
+        Assert.Equal(original.Truncated, back.Truncated);
+        var m = Assert.Single(back.Messages);
+        Assert.Equal(message.Id, m.Id);
+        Assert.Equal(message.Category, m.Category);
+        Assert.Equal(message.Target, m.Target);
+        Assert.Equal(message.PayloadJson, m.PayloadJson);
+        Assert.Equal(message.RetryCount, m.RetryCount);
+        Assert.Equal(message.MaxRetries, m.MaxRetries);
+        Assert.Equal(message.RetryIntervalMs, m.RetryIntervalMs);
+        Assert.Equal(message.CreatedAt, m.CreatedAt);
+        Assert.Equal(message.LastAttemptAt, m.LastAttemptAt);
+        Assert.Equal(message.Status, m.Status);
+        Assert.Equal(message.LastError, m.LastError);
+        Assert.Equal(message.OriginInstanceName, m.OriginInstanceName);
+        Assert.Equal(message.ExecutionId, m.ExecutionId);
+        Assert.Equal(message.SourceScript, m.SourceScript);
+        Assert.Equal(message.ParentExecutionId, m.ParentExecutionId);
+    }
+
+    [Fact]
+    public void SfBufferResyncAck_RoundTripsOnTheWire()
+    {
+        var original = new SfBufferResyncAck("resync-1", 42);
+
+        var back = RoundTrip(original);
+
+        Assert.Equal(original.ResyncId, back.ResyncId);
+        Assert.Equal(original.RowCount, back.RowCount);
+    }
+
+    // Type-identity pins: the reflective-JSON wire embeds CLR type manifests, so a
+    // rename/move of either type silently breaks resync across a rolling upgrade.
+    [Theory]
+    [InlineData(typeof(SfBufferSnapshotChunk), "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.SfBufferSnapshotChunk")]
+    [InlineData(typeof(SfBufferResyncAck), "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.SfBufferResyncAck")]
+    public void ResyncContract_TypeIdentity_IsPinned(Type type, string expectedFullName) =>
+        Assert.Equal(expectedFullName, type.FullName);
+}