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