diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs index 6e344bac..de61abef 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs @@ -39,6 +39,48 @@ public class SiteReplicationActor : ReceiveActor /// private const int MaxResyncRows = 10_000; + /// + /// Estimated per-chunk payload budget for a resync snapshot. Akka remoting's default + /// maximum-frame-size is 128 000 bytes and BuildHocon sets no override, + /// so the monolithic 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. + /// + internal const int MaxResyncChunkBytes = 64_000; + + /// Row cap per resync chunk (bounds a chunk even when every row is tiny). + internal const int MaxResyncChunkRows = 200; + + /// + /// 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). + /// + internal static List> ChunkForRemoting( + IReadOnlyList rows, int maxChunkBytes, int maxChunkRows) + { + var chunks = new List>(); + var current = new List(); + 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(); + currentBytes = 0; + } + current.Add(row); + currentBytes += estimate; + } + if (current.Count > 0) chunks.Add(current); + return chunks; + } + /// /// Initializes a new and registers Akka message handlers. /// @@ -111,7 +153,8 @@ public class SiteReplicationActor : ReceiveActor // Anti-entropy — full S&F buffer resync on peer (re)join Receive(HandleRequestSfBufferResync); - Receive(HandleSfBufferSnapshot); + Receive(HandleSfResyncSnapshotLoaded); + Receive(HandleSfBufferSnapshot); // legacy monolithic handler — retained for rolling compat } /// @@ -388,9 +431,10 @@ public class SiteReplicationActor : ReceiveActor /// /// 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. + /// with a sequence of byte-budgeted + /// s (up to 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. /// 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)); + } + + /// + /// 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. + /// + private void HandleSfResyncSnapshotLoaded(SfResyncSnapshotLoaded msg) + { + var chunks = ChunkForRemoting(msg.Messages, MaxResyncChunkBytes, MaxResyncChunkRows); + if (chunks.Count == 0) chunks.Add(new List()); // 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. } /// @@ -451,6 +519,10 @@ public class SiteReplicationActor : ReceiveActor _logger.LogError(t.Exception, "Failed to apply S&F buffer resync snapshot"); }); } + + /// Internal: the resync snapshot finished loading; chunk and send to the requester. + internal sealed record SfResyncSnapshotLoaded( + IActorRef ReplyTo, List Messages, bool Truncated); } /// @@ -467,3 +539,24 @@ public sealed record RequestSfBufferResync; /// Akka remoting; the message list rides the default serializer. /// public sealed record SfBufferSnapshot(List Messages, bool Truncated); + +/// +/// Active→standby: one sequenced chunk of a full-buffer anti-entropy snapshot +/// (review 02 round 2, N2 — the monolithic exceeds Akka +/// remoting's default 128 000-byte frame for any realistic backlog). All chunks of one +/// resync share ; is 1-based up to +/// . Additive message — the legacy monolithic snapshot +/// handler is retained for rolling upgrades. Crosses intra-site Akka remoting (NOT +/// ClusterClient — ClusterClientContractLockTests is intentionally not involved). +/// +public sealed record SfBufferSnapshotChunk( + string ResyncId, int Sequence, int TotalChunks, + List Messages, bool Truncated); + +/// +/// Standby→active: delivery confirmation — the standby assembled all chunks of +/// and applied them atomically ( +/// rows installed). Absence within the ack window is surfaced by the active node +/// (Warning + counter) — the silent-loss mode N2 flagged. +/// +public sealed record SfBufferResyncAck(string ResyncId, int RowCount); 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 7dccdddd..2ccd9ea4 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs +++ b/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(TimeSpan.FromSeconds(3)); - Assert.Single(snapshot.Messages); - Assert.False(snapshot.Truncated); + var chunk = ExpectMsg(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 + { 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(TimeSpan.FromSeconds(5)); + var rest = Enumerable.Range(1, first.TotalChunks - 1) + .Select(_ => ExpectMsg(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, }; + /// + /// Builds a resync-test message with a settable payload (additive to + /// — the chunker sizes on PayloadJson length). + /// + 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, + }; + + /// Constructs a with the given active-node check + /// (the resync chunk/ack tests Tell to and expect from ). + private IActorRef CreateResyncActor(Func isActive) => + ActorOf(Props.Create(() => new ResyncTestActor( + _storage, _sfStorage, _replicationService, SiteRole, + NullLogger.Instance, CreateTestProbe().Ref, isActive))); + /// Test message: drives directly, /// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer). private sealed record TriggerPeerTracked; diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs new file mode 100644 index 00000000..43560a14 --- /dev/null +++ b/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; + +/// +/// Characterization pin for the chunked anti-entropy resync contract (review 02 round 2, +/// N2). (active→standby) and +/// (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.) +/// +public class ResyncWireSerializationPinTests : TestKit +{ + private T RoundTrip(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 { 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); +}