feat(cluster): simultaneous-cold-start split-brain guard (opt-in)

Closes the residual Phase-7 gap: when BOTH nodes of a 2-node pair cold-start at the
same instant, each self-first seed runs FirstSeedNodeProcess, times out waiting for
the other, and forms its own 1-node cluster — two Primaries in one pair (the Phase 6
live gate reproduced this reliably on docker). The prior mitigation was operational
(staggered start / compose depends_on), which does not exist on production hardware.

The guard (dark switch Cluster:BootstrapGuard:Enabled, default OFF) prevents the split
without giving up cold-start-alone:
- The lower-address node is the preferred founder: self-first, forms immediately.
- The higher node probes its partner's Akka port (TCP connect) up to PartnerProbeSeconds:
  reachable => peer-first (join the founder, never race it); unreachable => self-first
  (partner is dead, form alone). The order is decided BEFORE the single JoinSeedNodes,
  from an explicit reachability signal — never re-formed mid-handshake (the retired
  SelfFormAfter failure mode). When on, Akka gets no config seeds (BuildClusterOptions)
  and ClusterBootstrapCoordinator drives the join.

Review-driven hardening: case-insensitive tie-break (a hostname-casing mismatch would
reopen the split); fail-fast validation of the timing knobs; the residual "founder dies
in the probe->join window" hang is documented and made operator-visible (warning + a
restart recovers). Real-ActorSystem coordinator tests cover the load-bearing higher-node
cold-start-alone case; 147/147 Cluster tests pass.

Live-gated on docker-dev (site-a = enablement demo, serialization removed, guard on;
site-b keeps depends_on-serialization + guard off as the A/B control): simultaneous
start -> site-a-1 founds, site-a-2 probes-reachable-joins -> 250/240, NO split;
higher-node-alone -> probes dead founder, self-forms -> 250; founder rejoin -> 240.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW

tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/AkkaClusterOptionsValidatorTests.cs

@@ -174,4 +174,44 @@ public sealed class AkkaClusterOptionsValidatorTests
 
         result.Failed.ShouldBeFalse(result.FailureMessage);
     }
+
+    /// <summary>The bootstrap-guard timing knobs default to positive values and pass when enabled.</summary>
+    [Fact]
+    public void Bootstrap_guard_defaults_pass()
+    {
+        var options = CentralNode("site-a-1", Seed("site-a-1"), Seed("site-a-2"));
+        options.BootstrapGuard = new ClusterBootstrapGuardOptions { Enabled = true };
+
+        var result = new AkkaClusterOptionsValidator().Validate(null, options);
+
+        result.Failed.ShouldBeFalse(result.FailureMessage);
+    }
+
+    /// <summary>
+    /// A zero <c>PartnerProbeSeconds</c> silently degrades the guard to "never wait, form alone
+    /// immediately" — re-opening the split. It must fail the host at boot, not run degraded.
+    /// </summary>
+    [Fact]
+    public void Bootstrap_guard_with_non_positive_probe_seconds_fails()
+    {
+        var options = CentralNode("site-a-1", Seed("site-a-1"), Seed("site-a-2"));
+        options.BootstrapGuard = new ClusterBootstrapGuardOptions { Enabled = true, PartnerProbeSeconds = 0 };
+
+        var result = new AkkaClusterOptionsValidator().Validate(null, options);
+
+        result.Failed.ShouldBeTrue();
+        result.FailureMessage.ShouldContain("PartnerProbeSeconds");
+    }
+
+    /// <summary>The timing knobs are NOT validated when the guard is off — a disabled guard is inert.</summary>
+    [Fact]
+    public void Bootstrap_guard_disabled_does_not_validate_timings()
+    {
+        var options = CentralNode("site-a-1", Seed("site-a-1"), Seed("site-a-2"));
+        options.BootstrapGuard = new ClusterBootstrapGuardOptions { Enabled = false, PartnerProbeSeconds = 0 };
+
+        var result = new AkkaClusterOptionsValidator().Validate(null, options);
+
+        result.Failed.ShouldBeFalse(result.FailureMessage);
+    }
 }

tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterBootstrapCoordinatorTests.cs

@@ -0,0 +1,204 @@
+using System.Net;
+using System.Net.Sockets;
+using Akka.Actor;
+using Akka.Cluster;
+using Akka.Hosting;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests;
+
+/// <summary>
+/// Real-ActorSystem tests for the simultaneous-cold-start split-brain guard
+/// (<see cref="ClusterBootstrapCoordinator"/> + <see cref="ClusterBootstrapGuard"/>), started through
+/// the SAME wiring as production (<see cref="ServiceCollectionExtensions.WithOtOpcUaClusterBootstrap"/>
+/// with <c>Cluster:BootstrapGuard:Enabled</c> + the coordinator hosted service). This subsystem has a
+/// history of bugs invisible to pure unit tests, so the load-bearing correctness properties are
+/// asserted against running nodes, mirroring <see cref="SelfFirstSeedBootstrapTests"/>.
+/// </summary>
+/// <remarks>
+/// Ephemeral loopback ports are 5-digit, so numeric order equals the guard's ordinal "host:port"
+/// string order — <c>Min(port)</c> is the founder, <c>Max(port)</c> is the higher (probing) node.
+/// </remarks>
+public sealed class ClusterBootstrapCoordinatorTests
+{
+    private static int FreePort()
+    {
+        var listener = new TcpListener(IPAddress.Loopback, 0);
+        listener.Start();
+        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+        listener.Stop();
+        return port;
+    }
+
+    /// <summary>
+    /// Starts a node with the bootstrap guard ENABLED, wired exactly as Program.cs does: empty Akka
+    /// seed list (via <c>BuildClusterOptions</c>) plus the coordinator hosted service that issues the
+    /// reachability-gated <c>JoinSeedNodes</c>. Seeds are listed self-first (as every shipped config
+    /// is); the guard, not the order, decides founder vs. joiner.
+    /// </summary>
+    private static async Task<IHost> StartGuardedNodeAsync(
+        string systemName, int selfPort, int peerPort, int probeSeconds = 4)
+    {
+        var self = $"akka.tcp://{systemName}@127.0.0.1:{selfPort}";
+        var peer = $"akka.tcp://{systemName}@127.0.0.1:{peerPort}";
+        var options = new AkkaClusterOptions
+        {
+            SystemName = systemName,
+            Hostname = "127.0.0.1",
+            PublicHostname = "127.0.0.1",
+            Port = selfPort,
+            Roles = new[] { "driver" },
+            SeedNodes = new[] { self, peer },
+            BootstrapGuard = new ClusterBootstrapGuardOptions
+            {
+                Enabled = true,
+                PartnerProbeSeconds = probeSeconds,
+                PartnerProbeIntervalMs = 250,
+                ProbeConnectTimeoutMs = 500,
+            },
+        };
+
+        var builder = Host.CreateDefaultBuilder();
+        builder.ConfigureServices(services =>
+        {
+            services.AddSingleton<IOptions<AkkaClusterOptions>>(Options.Create(options));
+            services.AddAkka(systemName, (ab, sp) => ab.WithOtOpcUaClusterBootstrap(sp));
+            // Mirror Program.cs — the coordinator is what drives the guarded join.
+            services.AddHostedService(sp => new ClusterBootstrapCoordinator(
+                () => sp.GetRequiredService<ActorSystem>(),
+                sp.GetRequiredService<IOptions<AkkaClusterOptions>>(),
+                sp.GetRequiredService<ILogger<ClusterBootstrapCoordinator>>()));
+        });
+
+        var host = builder.Build();
+        await host.StartAsync();
+        return host;
+    }
+
+    private static async Task<bool> WaitForUpMembersAsync(IHost host, int expected, TimeSpan timeout)
+    {
+        var cluster = Akka.Cluster.Cluster.Get(host.Services.GetRequiredService<ActorSystem>());
+        var deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            if (cluster.State.Members.Count(m => m.Status == MemberStatus.Up) >= expected) return true;
+            await Task.Delay(200);
+        }
+        return cluster.State.Members.Count(m => m.Status == MemberStatus.Up) >= expected;
+    }
+
+    private static async Task StopAsync(IHost host)
+    {
+        try { await host.StopAsync(); }
+        catch (Exception) { /* teardown only */ }
+        host.Dispose();
+    }
+
+    /// <summary>
+    /// The FOUNDER (lower address) with the guard on forms a cluster alone when its partner is dead —
+    /// exactly as un-guarded self-first does. The guard must not regress the preferred founder.
+    /// </summary>
+    [Fact]
+    public async Task Founder_forms_alone_when_partner_is_dead()
+    {
+        var low = FreePort();
+        var high = FreePort();
+        var founderPort = Math.Min(low, high);
+        var deadPeerPort = Math.Max(low, high); // nothing listening
+
+        var host = await StartGuardedNodeAsync("otopcua-guard-1", founderPort, deadPeerPort);
+        try
+        {
+            (await WaitForUpMembersAsync(host, 1, TimeSpan.FromSeconds(30)))
+                .ShouldBeTrue("the founder (lower address) must self-form immediately when its partner is down");
+        }
+        finally { await StopAsync(host); }
+    }
+
+    /// <summary>
+    /// THE LOAD-BEARING CASE. The HIGHER node with the guard on must still cold-start ALONE when its
+    /// partner is genuinely dead: it probes the (dead) founder, times out, and falls back to self-first.
+    /// If the guard's peer-first logic were wrong, this node would hang in JoinSeedNodeProcess forever —
+    /// the very regression a naive "always let the lower node found" rule would introduce.
+    /// </summary>
+    [Fact]
+    public async Task Higher_node_forms_alone_when_partner_is_dead_after_probing()
+    {
+        var low = FreePort();
+        var high = FreePort();
+        var higherPort = Math.Max(low, high);
+        var deadFounderPort = Math.Min(low, high); // the founder is dead
+
+        var host = await StartGuardedNodeAsync("otopcua-guard-2", higherPort, deadFounderPort, probeSeconds: 3);
+        try
+        {
+            // Must come Up AFTER the probe window (~3 s) expires and it falls back to self-first.
+            (await WaitForUpMembersAsync(host, 1, TimeSpan.FromSeconds(30)))
+                .ShouldBeTrue("the higher node must fall back to self-first and form alone once the dead partner never answers the probe");
+        }
+        finally { await StopAsync(host); }
+    }
+
+    /// <summary>
+    /// FALSIFIABILITY CONTROL for the test above: the higher node does NOT form alone during the probe
+    /// window — it is genuinely waiting/probing, not self-forming immediately (which would mean the
+    /// guard is inert). If this ever starts seeing a member before the window, the probe is not gating.
+    /// </summary>
+    [Fact]
+    public async Task Higher_node_does_not_form_before_the_probe_window_expires()
+    {
+        var low = FreePort();
+        var high = FreePort();
+        var higherPort = Math.Max(low, high);
+        var deadFounderPort = Math.Min(low, high);
+
+        var host = await StartGuardedNodeAsync("otopcua-guard-3", higherPort, deadFounderPort, probeSeconds: 10);
+        try
+        {
+            // Well within the 10 s probe window: still no cluster, because it is probing the dead founder.
+            (await WaitForUpMembersAsync(host, 1, TimeSpan.FromSeconds(3)))
+                .ShouldBeFalse("the higher node must probe for the founder, not self-form immediately");
+        }
+        finally { await StopAsync(host); }
+    }
+
+    /// <summary>
+    /// The higher node JOINS a live founder rather than forming a second cluster: start the founder,
+    /// then the higher node — its probe finds the founder reachable, it commits peer-first, and the
+    /// pair converges to ONE cluster of two.
+    /// </summary>
+    [Fact]
+    public async Task Higher_node_joins_a_live_founder()
+    {
+        const string systemName = "otopcua-guard-4";
+        var low = FreePort();
+        var high = FreePort();
+        var founderPort = Math.Min(low, high);
+        var higherPort = Math.Max(low, high);
+
+        var founder = await StartGuardedNodeAsync(systemName, founderPort, higherPort);
+        IHost? higher = null;
+        try
+        {
+            (await WaitForUpMembersAsync(founder, 1, TimeSpan.FromSeconds(30)))
+                .ShouldBeTrue("the founder must be Up before the higher node probes it");
+
+            higher = await StartGuardedNodeAsync(systemName, higherPort, founderPort);
+
+            (await WaitForUpMembersAsync(higher, 2, TimeSpan.FromSeconds(60)))
+                .ShouldBeTrue("the higher node must join the founder, forming one cluster of two");
+            (await WaitForUpMembersAsync(founder, 2, TimeSpan.FromSeconds(60)))
+                .ShouldBeTrue("the founder must see the higher node as a member of ITS cluster");
+        }
+        finally
+        {
+            if (higher is not null) await StopAsync(higher);
+            await StopAsync(founder);
+        }
+    }
+}

tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/ClusterBootstrapGuardTests.cs

@@ -0,0 +1,137 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Cluster;
+
+namespace ZB.MOM.WW.OtOpcUa.Cluster.Tests;
+
+/// <summary>
+/// Unit tests for <see cref="ClusterBootstrapGuard"/> — the pure decision core of the
+/// simultaneous-cold-start split-brain guard. The runtime probe + JoinSeedNodes wiring lives in
+/// <c>ClusterBootstrapCoordinator</c> and is covered by the live gate.
+/// </summary>
+public sealed class ClusterBootstrapGuardTests
+{
+    private const string A = "akka.tcp://otopcua@site-a-1:4053";
+    private const string B = "akka.tcp://otopcua@site-a-2:4053";
+
+    [Fact]
+    public void Lower_address_node_is_the_founder_and_needs_no_probe()
+    {
+        // site-a-1 < site-a-2, so on site-a-1 the guard makes it the founder.
+        var role = ClusterBootstrapGuard.Analyze(new[] { A, B }, "site-a-1", 4053);
+
+        role.Applies.ShouldBeTrue();
+        role.IsFounder.ShouldBeTrue();
+        role.SelfFirstOrder.ShouldBe(new[] { A, B });
+    }
+
+    [Fact]
+    public void Higher_address_node_is_not_the_founder_and_targets_the_partner_for_probing()
+    {
+        // On site-a-2, the partner is site-a-1 (the lower/founder).
+        var role = ClusterBootstrapGuard.Analyze(new[] { B, A }, "site-a-2", 4053);
+
+        role.Applies.ShouldBeTrue();
+        role.IsFounder.ShouldBeFalse();
+        role.PartnerHost.ShouldBe("site-a-1");
+        role.PartnerPort.ShouldBe(4053);
+    }
+
+    [Fact]
+    public void Tie_break_is_symmetric_regardless_of_seed_order()
+    {
+        // Both nodes must agree on who founds no matter how each lists its seeds.
+        var onLower = ClusterBootstrapGuard.Analyze(new[] { A, B }, "site-a-1", 4053);
+        var onHigher = ClusterBootstrapGuard.Analyze(new[] { B, A }, "site-a-2", 4053);
+
+        onLower.IsFounder.ShouldBeTrue();
+        onHigher.IsFounder.ShouldBeFalse();
+        // Exactly one founder.
+        (onLower.IsFounder ^ onHigher.IsFounder).ShouldBeTrue();
+    }
+
+    [Fact]
+    public void Higher_node_joins_the_founder_when_partner_is_reachable()
+    {
+        var role = ClusterBootstrapGuard.Analyze(new[] { B, A }, "site-a-2", 4053);
+
+        // Reachable ⇒ peer-first ⇒ JoinSeedNodeProcess ⇒ joins the founder, never self-forms.
+        ClusterBootstrapGuard.HigherNodeOrder(role, partnerReachable: true)
+            .ShouldBe(new[] { A, B }); // partner (site-a-1) first
+    }
+
+    [Fact]
+    public void Higher_node_forms_alone_when_partner_is_unreachable()
+    {
+        var role = ClusterBootstrapGuard.Analyze(new[] { B, A }, "site-a-2", 4053);
+
+        // Unreachable after the window ⇒ partner is dead ⇒ self-first ⇒ cold-start-alone preserved.
+        ClusterBootstrapGuard.HigherNodeOrder(role, partnerReachable: false)
+            .ShouldBe(new[] { B, A }); // self (site-a-2) first
+    }
+
+    [Fact]
+    public void Guard_does_not_apply_to_a_single_seed_node()
+    {
+        // A driver-only site node seeded solely by central-1 is not a pair seed — guard is inert.
+        var role = ClusterBootstrapGuard.Analyze(new[] { "akka.tcp://otopcua@central-1:4053" }, "site-x", 4053);
+
+        role.Applies.ShouldBeFalse();
+    }
+
+    [Fact]
+    public void Guard_does_not_apply_when_self_is_absent_from_the_two_seeds()
+    {
+        var role = ClusterBootstrapGuard.Analyze(new[] { A, B }, "central-1", 4053);
+
+        role.Applies.ShouldBeFalse();
+    }
+
+    [Fact]
+    public void Guard_does_not_apply_to_three_or_more_seeds()
+    {
+        var role = ClusterBootstrapGuard.Analyze(
+            new[] { A, B, "akka.tcp://otopcua@site-a-3:4053" }, "site-a-1", 4053);
+
+        role.Applies.ShouldBeFalse();
+    }
+
+    [Fact]
+    public void Guard_does_not_apply_when_a_seed_is_malformed()
+    {
+        var role = ClusterBootstrapGuard.Analyze(new[] { A, "not-a-seed" }, "site-a-1", 4053);
+
+        role.Applies.ShouldBeFalse();
+    }
+
+    [Theory]
+    [InlineData("akka.tcp://otopcua@site-a-1:4053", "site-a-1", 4053)]
+    [InlineData("akka.tcp://otopcua@10.0.0.5:4053/", "10.0.0.5", 4053)]
+    [InlineData("  akka.tcp://otopcua@host.lan:1234  ", "host.lan", 1234)]
+    public void TryParse_extracts_host_and_port(string seed, string expectedHost, int expectedPort)
+    {
+        ClusterBootstrapGuard.TryParse(seed, out var host, out var port).ShouldBeTrue();
+        host.ShouldBe(expectedHost);
+        port.ShouldBe(expectedPort);
+    }
+
+    [Theory]
+    [InlineData("")]
+    [InlineData("garbage")]
+    [InlineData("akka.tcp://otopcua@host")] // no port
+    public void TryParse_rejects_malformed_seeds(string seed)
+    {
+        ClusterBootstrapGuard.TryParse(seed, out _, out _).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void Port_participates_in_the_tie_break_when_hosts_are_equal()
+    {
+        // Shared-host loopback pair (test rig): same host, different ports — port breaks the tie.
+        const string low = "akka.tcp://otopcua@127.0.0.1:4053";
+        const string high = "akka.tcp://otopcua@127.0.0.1:4054";
+
+        ClusterBootstrapGuard.Analyze(new[] { low, high }, "127.0.0.1", 4053).IsFounder.ShouldBeTrue();
+        ClusterBootstrapGuard.Analyze(new[] { high, low }, "127.0.0.1", 4054).IsFounder.ShouldBeFalse();
+    }
+}