feat(cluster): InitJoin self-form fallback armed via Akka.Hosting startup task

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

src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ClusterBootstrapFallback.cs

@@ -0,0 +1,106 @@
+using Akka.Actor;
+using Microsoft.Extensions.Logging;
+
+namespace ZB.MOM.WW.OtOpcUa.Cluster;
+
+/// <summary>
+/// InitJoin self-form fallback (decision 2026-07-22, scadaproj/akka_failover.md §6.1).
+/// Akka only lets the FIRST listed seed form a NEW cluster; every other node retries InitJoin
+/// forever. So "both nodes are seed nodes" (<see cref="AkkaClusterOptions.SeedNodes"/>) does NOT
+/// mean either can cold-start alone — a non-first seed booting while its peer is down waits
+/// indefinitely ("auto-down removes the crash outage, not this one" — <c>docs/Redundancy.md</c>).
+/// This watchdog waits <see cref="AkkaClusterOptions.SelfFormAfter"/> for membership; on expiry it
+/// forms a cluster on itself.
+///
+/// <para><b>Island safety.</b> Fires ONLY when this node's own address is in its own seed list.
+/// A node that is not a seed (never legitimately first) must keep waiting: if it self-formed, a
+/// later-booting real seed would form a second cluster and the two can never merge. That is the
+/// current docker-dev site-node topology — site-a/site-b nodes list only central-1. For nodes that
+/// ARE seeds, sequential recovery is island-free — Akka's join protocol prefers an existing cluster
+/// (a booting node only self-forms when NO seed answers InitJoin), so a peer booting after this
+/// node self-formed simply joins it.</para>
+///
+/// <para><b>Races are benign.</b> If the join completes between window expiry and
+/// <c>Cluster.Join(SelfAddress)</c>, Akka ignores the join — a node joins a cluster at most once
+/// per incarnation. The residual risk is both pair nodes cold-starting inside the window while
+/// mutually unreachable (a boot-time partition): both self-form, the same dual-active class the
+/// auto-down downing strategy already accepts, with the same recovery (restart one side).</para>
+///
+/// <para>Deliberately duplicated from ScadaBridge's equivalent (like the termination watchdogs) —
+/// the two repos share the behavior, not a package.</para>
+/// </summary>
+public static class ClusterBootstrapFallback
+{
+    /// <summary>
+    /// Arms the fallback on a freshly created actor system. Safe to call unconditionally: it no-ops
+    /// (with an explanatory log) when the fallback is disabled or when this node is not one of its
+    /// own seeds.
+    /// </summary>
+    /// <param name="system">The actor system whose cluster membership is being watched.</param>
+    /// <param name="options">The bound cluster options carrying the window and the seed list.</param>
+    /// <param name="logger">Logger for the armed / inert / self-forming decisions.</param>
+    public static void Arm(ActorSystem system, AkkaClusterOptions options, ILogger logger)
+    {
+        ArgumentNullException.ThrowIfNull(system);
+        ArgumentNullException.ThrowIfNull(options);
+        ArgumentNullException.ThrowIfNull(logger);
+
+        if (options.SelfFormAfter is not { } window || window <= TimeSpan.Zero)
+        {
+            logger.LogInformation(
+                "Cluster self-form fallback disabled (SelfFormAfter not set) — a node cold-starting "
+                + "while its peer is down will wait on InitJoin indefinitely.");
+            return;
+        }
+
+        var cluster = Akka.Cluster.Cluster.Get(system);
+        var self = cluster.SelfAddress;
+        var isSeed = options.SeedNodes.Any(s => TryParseAddress(s, out var a) && a.Equals(self));
+        if (!isSeed)
+        {
+            logger.LogInformation(
+                "Cluster self-form fallback inactive: this node ({Self}) is not in its own seed list "
+                + "[{Seeds}] — self-forming here would island it from the real seeds.",
+                self,
+                string.Join(", ", options.SeedNodes));
+            return;
+        }
+
+        var joined = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        cluster.RegisterOnMemberUp(() => joined.TrySetResult());
+
+        _ = Task.Run(async () =>
+        {
+            var winner = await Task.WhenAny(joined.Task, Task.Delay(window));
+            if (winner == joined.Task || system.WhenTerminated.IsCompleted)
+            {
+                return;
+            }
+
+            logger.LogWarning(
+                "No cluster membership after {Window} — no seed answered InitJoin (peer down at boot). "
+                + "Self-forming a cluster at {Self} so this node becomes operational; if the peer was "
+                + "merely partitioned (not dead), the pair is now dual-active — restart one side after "
+                + "the partition heals (accepted availability-first trade, decision 2026-07-22).",
+                window,
+                self);
+            cluster.Join(self);
+        });
+    }
+
+    private static bool TryParseAddress(string seed, out Address address)
+    {
+        try
+        {
+            address = Address.Parse(seed);
+            return true;
+        }
+        catch (Exception)
+        {
+            // A malformed seed entry must not disarm the fallback; bad seed URIs surface from
+            // Akka's own bootstrap.
+            address = Address.AllSystems;
+            return false;
+        }
+    }
+}

src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ServiceCollectionExtensions.cs

@@ -92,6 +92,23 @@ public static class ServiceCollectionExtensions
         // ActorSystem for exactly that reason: the effective value is the only one worth asserting.
         builder.AddHocon(BuildDowningHocon(options), HoconAddMode.Prepend);
 
+        // InitJoin self-form fallback (decision 2026-07-22): registered as an Akka.Hosting startup
+        // task so every host built through this bootstrap — production AND the test hosts in
+        // Cluster.Tests — arms it identically. Guarded inside Arm: disabled when SelfFormAfter is
+        // unset; inert when this node is not in its own seed list (site nodes seeded only by
+        // central-1 must wait, not island).
+        // ILoggerFactory is fully qualified: `using Microsoft.Extensions.Logging` would make the
+        // LogLevel in ConfigureLoggers above ambiguous with Akka.Event.LogLevel.
+        var fallbackLogger =
+            (serviceProvider.GetService<Microsoft.Extensions.Logging.ILoggerFactory>()
+             ?? Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance)
+            .CreateLogger("ZB.MOM.WW.OtOpcUa.Cluster.ClusterBootstrapFallback");
+        builder.AddStartup((system, _) =>
+        {
+            ClusterBootstrapFallback.Arm(system, options, fallbackLogger);
+            return Task.CompletedTask;
+        });
+
         return builder;
     }