test(saf): failing two-node repro — leader-vs-oldest resync divergence wipes delivering node buffer (plan R2-02 T2)

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+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<StoreAndForwardStorage>.Instance);
+        await sfStorage.InitializeAsync();
+        var siteStorage = new SiteStorageService($"Data Source={siteDb}",
+            NullLogger<SiteStorageService>.Instance);
+        var replicationService = new ReplicationService(
+            new StoreAndForwardOptions(), NullLogger<ReplicationService>.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<SiteReplicationActor>.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<Task<bool>> 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);
+    }
+}

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<TwoNodeClusterFixture> 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();