From 6e0fa21307e3812d7da2b3d49fddb7478cca9cd4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 09:46:16 -0400 Subject: [PATCH] =?UTF-8?q?test(saf):=20failing=20two-node=20repro=20?= =?UTF-8?q?=E2=80=94=20leader-vs-oldest=20resync=20divergence=20wipes=20de?= =?UTF-8?q?livering=20node=20buffer=20(plan=20R2-02=20T2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Cluster/SfBufferResyncPredicateTests.cs | 94 +++++++++++++++++++ .../Cluster/TwoNodeClusterFixture.cs | 9 +- 2 files changed, 99 insertions(+), 4 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs new file mode 100644 index 00000000..0dac6c10 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs @@ -0,0 +1,94 @@ +using Akka.Actor; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; +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.IntegrationTests.Cluster; + +/// +/// N1 regression (review 02 round 2, Critical): the resync authority must use the same +/// oldest-Up predicate as the S&F delivery gate. Divergence scenario = the delivering node +/// is OLDEST but not LEADER (leader = lowest address), the exact state a rolling restart of +/// the lower-address node produces. Pre-fix the delivering node requests a resync from the +/// stale peer and ReplaceAllAsync wipes its live buffer. +/// +public class SfBufferResyncPredicateTests +{ + [Fact] + public async Task OldestButNotLeaderNode_KeepsItsBuffer_AndSeedsTheJoiner() + { + // Two explicit ports, deliberately assigned so the FIRST-started (oldest, + // delivering) node has the HIGHER address → the second node is cluster leader. + var p1 = TwoNodeClusterFixture.GetFreeTcpPort(); + var p2 = TwoNodeClusterFixture.GetFreeTcpPort(); + var (portHigh, portLow) = p1 > p2 ? (p1, p2) : (p2, p1); + + await using var fixture = await TwoNodeClusterFixture.StartAsync( + role: "site-int", portA: portHigh, portB: portLow); + + // Real S&F storage + replication actor per node, production default predicate + // (no isActiveOverride) — the exact wiring under test. + var (storageOldest, _) = await CreateReplicationActorAsync(fixture.NodeA, "oldest"); + var (storageJoiner, _) = await CreateReplicationActorAsync(fixture.NodeB, "joiner"); + + // The delivering (oldest) node has a live buffered row the standby never saw. + await storageOldest.EnqueueAsync(NewMessage("live-row")); + + // Trigger peer (re)tracking on both sides: each actor got InitialStateAsSnapshot + // in PreStart, but the enqueue raced it — re-deliver via a fresh MemberUp is not + // needed; OnPeerTracked already fired on join. The resync exchange is async: + // wait until the JOINER holds the row (proves the snapshot flowed oldest→joiner, + // the correct direction). Pre-fix this times out (the joiner, as leader, never + // requests) AND the oldest node's row is deleted by the stale wipe. + await AwaitAsync(async () => await storageJoiner.GetMessageByIdAsync("live-row") != null, + TimeSpan.FromSeconds(20), + "joiner never received the resync snapshot (resync ran in the wrong direction)"); + + // And the delivering node's buffer is untouched — the N1 wipe assertion. + Assert.NotNull(await storageOldest.GetMessageByIdAsync("live-row")); + } + + private static async Task<(StoreAndForwardStorage Storage, IActorRef Actor)> CreateReplicationActorAsync( + ActorSystem node, string tag) + { + var sfDb = Path.Combine(Path.GetTempPath(), $"sf-resync-{tag}-{Guid.NewGuid():N}.db"); + var siteDb = Path.Combine(Path.GetTempPath(), $"site-resync-{tag}-{Guid.NewGuid():N}.db"); + var sfStorage = new StoreAndForwardStorage($"Data Source={sfDb}", + NullLogger.Instance); + await sfStorage.InitializeAsync(); + var siteStorage = new SiteStorageService($"Data Source={siteDb}", + NullLogger.Instance); + var replicationService = new ReplicationService( + new StoreAndForwardOptions(), NullLogger.Instance); + // Name MUST be "site-replication" — SendToPeer targets /user/site-replication. + var actor = node.ActorOf(Props.Create(() => new SiteReplicationActor( + siteStorage, sfStorage, replicationService, "site-int", + NullLogger.Instance, null, null, null, null)), + "site-replication"); + return (sfStorage, actor); + } + + private static StoreAndForwardMessage NewMessage(string id) => new() + { + Id = id, + Category = StoreAndForwardCategory.Notification, + Target = "central", + PayloadJson = "{}", + CreatedAt = DateTimeOffset.UtcNow, + Status = StoreAndForwardMessageStatus.Pending, + MaxRetries = 0, + }; + + private static async Task AwaitAsync(Func> condition, TimeSpan timeout, string why) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (await condition()) return; + await Task.Delay(250); + } + throw new TimeoutException(why); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs index ede98c32..3d1c94da 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs @@ -22,11 +22,12 @@ public sealed class TwoNodeClusterFixture : IAsyncDisposable public int PortB { get; private set; } public static async Task StartAsync( - string role = "Central", TimeSpan? stableAfter = null) + string role = "Central", TimeSpan? stableAfter = null, + int? portA = null, int? portB = null) { var f = new TwoNodeClusterFixture(); - f.PortA = GetFreeTcpPort(); - f.PortB = GetFreeTcpPort(); + f.PortA = portA ?? GetFreeTcpPort(); + f.PortB = portB ?? GetFreeTcpPort(); f.NodeA = f.StartNode(f.PortA, role, stableAfter); await WaitForMembersUp(f.NodeA, 1, TimeSpan.FromSeconds(20)); f.NodeB = f.StartNode(f.PortB, role, stableAfter); @@ -98,7 +99,7 @@ public sealed class TwoNodeClusterFixture : IAsyncDisposable throw new TimeoutException($"Member {removed} was never removed from {cluster.SelfAddress}'s view — SBR did not down it."); } - private static int GetFreeTcpPort() + public static int GetFreeTcpPort() { var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0); listener.Start();