feat(cluster): opt-in leader-auto-down strategy - in-repo port of Akka AutoDowning (ScadaBridge#106)

Adds LeaderAutoDowningProvider (Host/Actors), a line-for-line port of Akka.NET
1.5.62 AutoDowning/AutoDown/AutoDownBase read from the decompiled assembly: the
leader among the reachable members downs a peer unreachable for StableAfter; a
non-leader records it as pending and downs it on becoming leader; reachable-again
or removed members are forgotten; Down/Exiting members are skipped. Its window is
carried under scadabridge.cluster.leader-auto-down.unreachable-after, so no
deprecated auto-down-unreachable-after key is emitted and Akka's boot-time
"keep-majority will be used instead" warning goes away. DownRemovalMargin reads
akka.cluster.down-removal-margin exactly as AutoDowning does (default off).

New SplitBrainResolverStrategy value "leader-auto-down", wired through BuildHocon
and ClusterOptionsValidator. The default stays "auto-down"; auto-down and
keep-oldest HOCON is byte-identical to before. Placed in Host, not
ClusterInfrastructure, per the Code Placement rule (ClusterInfrastructure stays
Akka-free).

Tests: HOCON + validator units; LeaderAutoDownTests ports Akka's AutoDownSpec
cases (leader-only, pending-on-leader-change, losing leadership, reachable again,
removed, Down/Exiting skipped, single Down per member); LeaderAutoDownFailoverTests
clones SbrFailoverTests' oldest-crash and younger-crash cases on a real two-node
cluster plus a test that Akka actually loaded the provider and its downing actor.
Mutation-checked: a provider that never calls Cluster.Down fails both failover
tests; never-downs, leader-check-dropped, reachable-ignored and pending-not-flushed
mutations each fail the unit tests.

Remaining gate before switching the default: the docker failover drill
(DRILL_MODE=active and standby mode) on leader-auto-down.

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

src/ZB.MOM.WW.ScadaBridge.ClusterInfrastructure/ClusterOptions.cs

@@ -50,7 +50,7 @@ public class ClusterOptions
     public List<string> SeedNodes { get; set; } = new();
 
     /// <summary>
-    /// Downing strategy for unreachable members. Two supported values:
+    /// Downing strategy for unreachable members. Three supported values:
     /// <list type="bullet">
     /// <item><c>auto-down</c> (default, decision 2026-07-21) — availability-first: each
     /// side downs the unreachable peer after <see cref="StableAfter"/>, so a hard crash
@@ -59,6 +59,12 @@ public class ClusterOptions
     /// an operator restarts one side. Chosen because ScadaBridge pairs run one node per
     /// VM with no shared lease infrastructure, and a stalled system is a bigger risk
     /// than a rare partition.</item>
+    /// <item><c>leader-auto-down</c> (opt-in, ScadaBridge#106) — identical semantics to
+    /// <c>auto-down</c>, implemented by the in-repo <c>LeaderAutoDowningProvider</c> (Host) instead
+    /// of Akka's deprecated <c>AutoDowning</c>, so it survives that provider's removal from a
+    /// future Akka.NET and does not log the boot-time deprecation warning. Not yet the default:
+    /// the docker failover drill must pass on it first
+    /// (<c>docs/plans/2026-11-09-autodowning-replacement-plan.md</c>).</item>
     /// <item><c>keep-oldest</c> — partition-safe SBR: downs the side without the oldest
     /// member. In a TWO-node cluster this makes a crash of the oldest/active node a
     /// total outage: Akka's <c>down-if-alone</c> only rescues the survivor when its own

src/ZB.MOM.WW.ScadaBridge.ClusterInfrastructure/ClusterOptionsValidator.cs

@@ -20,10 +20,13 @@ public sealed class ClusterOptionsValidator : OptionsValidatorBase<ClusterOption
     /// <c>static-quorum</c> quorum-size 1 trips Akka's IsTooManyMembers guard (DownAll
     /// on any unreachability in a 2-node cluster) and <c>keep-majority</c> keys the
     /// fatal crash to the lowest-address node instead of the oldest.
+    /// <c>leader-auto-down</c> (ScadaBridge#106) is the in-repo port of <c>auto-down</c>
+    /// (same semantics, no dependency on Akka's deprecated <c>AutoDowning</c>).
     /// </summary>
     private static readonly HashSet<string> AllowedStrategies = new(StringComparer.OrdinalIgnoreCase)
     {
         "auto-down",
+        "leader-auto-down",
         "keep-oldest"
     };
 
@@ -48,7 +51,7 @@ public sealed class ClusterOptionsValidator : OptionsValidatorBase<ClusterOption
         builder.RequireThat(
             !string.IsNullOrWhiteSpace(options.SplitBrainResolverStrategy)
             && AllowedStrategies.Contains(options.SplitBrainResolverStrategy),
-            $"ClusterOptions.SplitBrainResolverStrategy must be 'auto-down' or 'keep-oldest' for a " +
+            $"ClusterOptions.SplitBrainResolverStrategy must be 'auto-down', 'leader-auto-down' or 'keep-oldest' for a " +
             $"two-node cluster; '{options.SplitBrainResolverStrategy}' would risk a total cluster " +
             "shutdown on a partition or an unreachability event.");
 
@@ -70,7 +73,7 @@ public sealed class ClusterOptionsValidator : OptionsValidatorBase<ClusterOption
             $"FailureDetectionThreshold ({options.FailureDetectionThreshold}); otherwise nodes are " +
             "declared unreachable before a heartbeat can arrive.");
 
-        // DownIfAlone is a keep-oldest knob; under auto-down each side downs the
+        // DownIfAlone is a keep-oldest knob; under (leader-)auto-down each side downs the
         // unreachable peer regardless, so the flag is inert and any value is fine.
         var isKeepOldest = string.Equals(
             options.SplitBrainResolverStrategy, "keep-oldest", StringComparison.OrdinalIgnoreCase);

src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs

@@ -235,7 +235,11 @@ public class AkkaHostedService : IHostedService
     /// <see cref="ClusterOptions.StableAfter"/> — the leader among the REACHABLE members
     /// downs the unreachable peer, so a crash of either node (oldest included) fails
     /// over to the survivor; the accepted trade is dual-active during a real network
-    /// partition. Any other value takes the SBR path, where the
+    /// partition. <c>leader-auto-down</c> (ScadaBridge#106, opt-in) installs the in-repo
+    /// <see cref="LeaderAutoDowningProvider"/> — the same semantics without Akka's deprecated
+    /// provider — with its duration at
+    /// <see cref="LeaderAutoDowningProvider.UnreachableAfterPath"/>.
+    /// Any other value takes the SBR path, where the
     /// <c>keep-oldest down-if-alone</c> flag is emitted from
     /// <see cref="ClusterOptions.DownIfAlone"/> rather than hard-coded, so the bound
     /// configuration value is actually consumed.
@@ -280,10 +284,17 @@ public class AkkaHostedService : IHostedService
         // accepted trade — decision 2026-07-21). Anything else: the SBR provider with
         // the configured active-strategy (keep-oldest), which is partition-safe but
         // cannot survive a crash of the oldest node in a two-node cluster.
-        var downingBlock = string.Equals(
-            clusterOptions.SplitBrainResolverStrategy, "auto-down", StringComparison.OrdinalIgnoreCase)
+        //
+        // leader-auto-down (ScadaBridge#106, opt-in): the in-repo LeaderAutoDowningProvider — a
+        // port of AutoDowning with identical semantics that survives AutoDowning's eventual
+        // removal from Akka.NET. Its duration lives under our own key, so no
+        // auto-down-unreachable-after is emitted (which also silences Akka's deprecation warning).
+        var strategy = clusterOptions.SplitBrainResolverStrategy;
+        var downingBlock = string.Equals(strategy, "auto-down", StringComparison.OrdinalIgnoreCase)
             ? $@"downing-provider-class = ""Akka.Cluster.AutoDowning, Akka.Cluster""
         auto-down-unreachable-after = {DurationHocon(clusterOptions.StableAfter)}"
+            : string.Equals(strategy, "leader-auto-down", StringComparison.OrdinalIgnoreCase)
+            ? $@"downing-provider-class = {QuoteHocon(LeaderAutoDowningProvider.ProviderClassName)}"
             : $@"downing-provider-class = ""Akka.Cluster.SBR.SplitBrainResolverProvider, Akka.Cluster""
         split-brain-resolver {{
             active-strategy = {QuoteHocon(clusterOptions.SplitBrainResolverStrategy)}
@@ -293,8 +304,15 @@ public class AkkaHostedService : IHostedService
             }}
         }}";
 
+        // The leader-auto-down duration sits outside akka.cluster so Akka never mistakes it
+        // for one of its own (deprecated) auto-down keys. Only emitted for that strategy, so
+        // the auto-down / keep-oldest documents are byte-identical to before.
+        var leaderAutoDownBlock = string.Equals(strategy, "leader-auto-down", StringComparison.OrdinalIgnoreCase)
+            ? $"{LeaderAutoDowningProvider.UnreachableAfterPath} = {DurationHocon(clusterOptions.StableAfter)}\n"
+            : string.Empty;
+
         return $@"
-audit-telemetry-dispatcher {{
+{leaderAutoDownBlock}audit-telemetry-dispatcher {{
     type = ForkJoinDispatcher
     throughput = 100
     dedicated-thread-pool {{

src/ZB.MOM.WW.ScadaBridge.Host/Actors/LeaderAutoDowningProvider.cs

@@ -0,0 +1,318 @@
+using System.Collections.Immutable;
+using Akka.Actor;
+using Akka.Cluster;
+using Akka.Configuration;
+using Akka.Event;
+
+namespace ZB.MOM.WW.ScadaBridge.Host.Actors;
+
+/// <summary>
+/// In-repo replacement for Akka.NET's deprecated <c>Akka.Cluster.AutoDowning</c> provider
+/// (ScadaBridge#51 plan, implemented by ScadaBridge#106). Selected by
+/// <c>ScadaBridge:Cluster:SplitBrainResolverStrategy = "leader-auto-down"</c>; the HOCON is
+/// emitted by <see cref="AkkaHostedService.BuildHocon"/>.
+/// <para>
+/// <b>Behaviour is a line-for-line port of Akka.NET 1.5.62 <c>AutoDowning</c> /
+/// <c>AutoDown</c> / <c>AutoDownBase</c></b> (read from the decompiled 1.5.62 assembly), so
+/// no operational contract changes when a pair switches from <c>auto-down</c>: the leader
+/// among the REACHABLE members downs a peer that has stayed unreachable for
+/// <c>unreachable-after</c> (= <c>ClusterOptions.StableAfter</c>); a non-leader records the
+/// peer as pending and downs it if it later becomes leader; a peer that becomes reachable
+/// again, or is removed, before the timer fires is forgotten. Because Akka computes the
+/// leader over reachable members only, a crash of EITHER node of a pair (oldest included)
+/// makes the survivor leader and it downs the crashed peer — the availability-first posture
+/// of the 2026-07-21 decision, with dual-active on a real partition as the accepted trade.
+/// </para>
+/// <para>
+/// Differences from <c>AutoDowning</c>, all deliberate and none behavioural:
+/// the duration is read from <see cref="UnreachableAfterPath"/> rather than the deprecated
+/// <c>akka.cluster.auto-down-unreachable-after</c> key (so Akka's boot-time deprecation
+/// warning — "keep-majority will be used instead" — is no longer logged), and the
+/// <see cref="DownRemovalMargin"/> is read from <c>akka.cluster.down-removal-margin</c>
+/// directly because <c>ClusterSettings.DownRemovalMargin</c> is internal (same off-switch
+/// parsing; default <c>off</c> = zero, exactly what AutoDowning returns).
+/// </para>
+/// <para>
+/// Akka instantiates this by reflection (<c>Akka.Cluster.DowningProvider.Load</c>) with the
+/// <c>(ActorSystem, Cluster)</c> constructor; the downing actor is created as
+/// <c>/system/cluster/core/daemon/downingProvider</c>.
+/// </para>
+/// </summary>
+public sealed class LeaderAutoDowningProvider : IDowningProvider
+{
+    /// <summary>HOCON path carrying the unreachable-before-down duration.</summary>
+    public const string UnreachableAfterPath = "scadabridge.cluster.leader-auto-down.unreachable-after";
+
+    /// <summary>Assembly-qualified type name for <c>akka.cluster.downing-provider-class</c>.</summary>
+    public static readonly string ProviderClassName =
+        $"{typeof(LeaderAutoDowningProvider).FullName}, {typeof(LeaderAutoDowningProvider).Assembly.GetName().Name}";
+
+    private readonly ActorSystem _system;
+    private readonly Akka.Cluster.Cluster _cluster;
+
+    /// <summary>Creates the provider; called by Akka via reflection.</summary>
+    /// <param name="system">The actor system the cluster runs in.</param>
+    /// <param name="cluster">The cluster extension being started.</param>
+    public LeaderAutoDowningProvider(ActorSystem system, Akka.Cluster.Cluster cluster)
+    {
+        _system = system;
+        _cluster = cluster;
+    }
+
+    /// <inheritdoc />
+    public TimeSpan DownRemovalMargin
+    {
+        get
+        {
+            var clusterConfig = _system.Settings.Config.GetConfig("akka.cluster");
+            var raw = clusterConfig.GetString("down-removal-margin", string.Empty).ToLowerInvariant();
+            return raw is "off" or "false" or "no"
+                ? TimeSpan.Zero
+                : clusterConfig.GetTimeSpan("down-removal-margin", null, true);
+        }
+    }
+
+    /// <inheritdoc />
+    /// <exception cref="ConfigurationException">
+    /// <see cref="UnreachableAfterPath"/> is missing or <c>off</c> — mirrors AutoDowning's
+    /// refusal to start without <c>auto-down-unreachable-after</c>.
+    /// </exception>
+    public Props DowningActorProps
+    {
+        get
+        {
+            var after = ReadUnreachableAfter(_system.Settings.Config);
+            if (!after.HasValue)
+            {
+                throw new ConfigurationException(
+                    $"LeaderAutoDowningProvider selected but '{UnreachableAfterPath}' not set");
+            }
+
+            return LeaderAutoDown.Props(after.Value, _cluster);
+        }
+    }
+
+    /// <summary>
+    /// Reads <see cref="UnreachableAfterPath"/> with Akka's off-switch semantics (Akka's own
+    /// <c>GetTimeSpanWithOffSwitch</c> is internal): missing or <c>off</c>/<c>false</c>/<c>no</c>
+    /// yields <see langword="null"/>.
+    /// </summary>
+    private static TimeSpan? ReadUnreachableAfter(Config config)
+    {
+        if (!config.HasPath(UnreachableAfterPath))
+        {
+            return null;
+        }
+
+        var raw = config.GetString(UnreachableAfterPath, string.Empty).ToLowerInvariant();
+        return raw is "off" or "false" or "no"
+            ? null
+            : config.GetTimeSpan(UnreachableAfterPath, null, true);
+    }
+}
+
+/// <summary>
+/// The downing actor for <see cref="LeaderAutoDowningProvider"/> — port of Akka's internal
+/// <c>AutoDown</c>: subscribes to cluster domain events and downs through
+/// <see cref="Akka.Cluster.Cluster.Down"/>.
+/// </summary>
+internal sealed class LeaderAutoDown : LeaderAutoDownBase
+{
+    private readonly Akka.Cluster.Cluster _cluster;
+    private readonly ILoggingAdapter _log = Context.GetLogger();
+
+    /// <summary>Props for the downing actor.</summary>
+    /// <param name="unreachableAfter">How long a member must stay unreachable before it is downed.</param>
+    /// <param name="cluster">The cluster extension.</param>
+    public static Props Props(TimeSpan unreachableAfter, Akka.Cluster.Cluster cluster) =>
+        Akka.Actor.Props.Create(() => new LeaderAutoDown(unreachableAfter, cluster));
+
+    /// <summary>Creates the actor.</summary>
+    /// <param name="unreachableAfter">How long a member must stay unreachable before it is downed.</param>
+    /// <param name="cluster">The cluster extension.</param>
+    public LeaderAutoDown(TimeSpan unreachableAfter, Akka.Cluster.Cluster cluster)
+        : base(unreachableAfter)
+    {
+        _cluster = cluster;
+    }
+
+    /// <inheritdoc />
+    public override Address SelfAddress => _cluster.SelfAddress;
+
+    /// <inheritdoc />
+    public override IScheduler Scheduler => Context.System.Scheduler;
+
+    /// <inheritdoc />
+    protected override void PreStart()
+    {
+        _cluster.Subscribe(Self, typeof(ClusterEvent.IClusterDomainEvent));
+        base.PreStart();
+    }
+
+    /// <inheritdoc />
+    protected override void PostStop()
+    {
+        _cluster.Unsubscribe(Self);
+        base.PostStop();
+    }
+
+    /// <inheritdoc />
+    /// <exception cref="InvalidOperationException">Called on a non-leader (a logic error).</exception>
+    public override void Down(Address node)
+    {
+        if (!IsLeader)
+        {
+            throw new InvalidOperationException("Must be leader to down node");
+        }
+
+        _log.Info("Leader is auto-downing unreachable node [{0}]", node);
+        _cluster.Down(node);
+    }
+}
+
+/// <summary>
+/// The cluster-free decision logic of <see cref="LeaderAutoDown"/> — port of Akka's internal
+/// <c>AutoDownBase</c>, split out (as Akka does) so the leader / reachability semantics can be
+/// unit tested by feeding cluster events directly.
+/// </summary>
+internal abstract class LeaderAutoDownBase : UntypedActor
+{
+    /// <summary>Timer message: <see cref="Node"/> has been unreachable for the full window.</summary>
+    /// <param name="Node">The member that was scheduled for downing.</param>
+    internal sealed record UnreachableTimeout(UniqueAddress Node);
+
+    /// <summary>
+    /// Members in these states are not scheduled — Akka's
+    /// <c>MembershipState.ConvergenceSkipUnreachableWithMemberStatus</c> (Down, Exiting).
+    /// </summary>
+    private static readonly ImmutableHashSet<MemberStatus> SkipMemberStatus =
+        ImmutableHashSet.Create(MemberStatus.Down, MemberStatus.Exiting);
+
+    private readonly TimeSpan _unreachableAfter;
+    private ImmutableDictionary<UniqueAddress, ICancelable> _scheduledUnreachable =
+        ImmutableDictionary<UniqueAddress, ICancelable>.Empty;
+    private ImmutableHashSet<UniqueAddress> _pendingUnreachable = ImmutableHashSet<UniqueAddress>.Empty;
+
+    /// <summary>Creates the base logic.</summary>
+    /// <param name="unreachableAfter">How long a member must stay unreachable before it is downed.</param>
+    protected LeaderAutoDownBase(TimeSpan unreachableAfter)
+    {
+        _unreachableAfter = unreachableAfter;
+    }
+
+    /// <summary>Whether this node is currently the cluster leader.</summary>
+    protected bool IsLeader { get; private set; }
+
+    /// <summary>This node's address, compared against the leader address.</summary>
+    public abstract Address SelfAddress { get; }
+
+    /// <summary>The scheduler used for the unreachable timers.</summary>
+    public abstract IScheduler Scheduler { get; }
+
+    /// <summary>Downs <paramref name="node"/>; only ever called while leader.</summary>
+    /// <param name="node">The member to down.</param>
+    public abstract void Down(Address node);
+
+    /// <inheritdoc />
+    protected override void PostStop()
+    {
+        foreach (var cancelable in _scheduledUnreachable.Values)
+        {
+            cancelable.Cancel();
+        }
+    }
+
+    /// <inheritdoc />
+    protected override void OnReceive(object message)
+    {
+        switch (message)
+        {
+            case ClusterEvent.CurrentClusterState state:
+                IsLeader = state.Leader != null && state.Leader.Equals(SelfAddress);
+                foreach (var member in state.Unreachable)
+                {
+                    UnreachableMember(member);
+                }
+                break;
+
+            case ClusterEvent.UnreachableMember unreachable:
+                UnreachableMember(unreachable.Member);
+                break;
+
+            case ClusterEvent.ReachableMember reachable:
+                Remove(reachable.Member.UniqueAddress);
+                break;
+
+            case ClusterEvent.MemberRemoved removed:
+                Remove(removed.Member.UniqueAddress);
+                break;
+
+            case ClusterEvent.LeaderChanged leaderChanged:
+                IsLeader = leaderChanged.Leader != null && leaderChanged.Leader.Equals(SelfAddress);
+                if (IsLeader)
+                {
+                    foreach (var node in _pendingUnreachable)
+                    {
+                        Down(node.Address);
+                    }
+
+                    _pendingUnreachable = ImmutableHashSet<UniqueAddress>.Empty;
+                }
+                break;
+
+            case UnreachableTimeout timeout:
+                if (_scheduledUnreachable.ContainsKey(timeout.Node))
+                {
+                    _scheduledUnreachable = _scheduledUnreachable.Remove(timeout.Node);
+                    DownOrAddPending(timeout.Node);
+                }
+                break;
+
+            // Every other IClusterDomainEvent is ignored, exactly as AutoDownBase does.
+        }
+    }
+
+    private void UnreachableMember(Member member)
+    {
+        if (!SkipMemberStatus.Contains(member.Status) && !_scheduledUnreachable.ContainsKey(member.UniqueAddress))
+        {
+            ScheduleUnreachable(member.UniqueAddress);
+        }
+    }
+
+    private void ScheduleUnreachable(UniqueAddress node)
+    {
+        if (_unreachableAfter == TimeSpan.Zero)
+        {
+            DownOrAddPending(node);
+            return;
+        }
+
+        var cancelable = Scheduler.ScheduleTellOnceCancelable(
+            _unreachableAfter, Self, new UnreachableTimeout(node), Self);
+        _scheduledUnreachable = _scheduledUnreachable.Add(node, cancelable);
+    }
+
+    private void DownOrAddPending(UniqueAddress node)
+    {
+        if (IsLeader)
+        {
+            Down(node.Address);
+        }
+        else
+        {
+            _pendingUnreachable = _pendingUnreachable.Add(node);
+        }
+    }
+
+    private void Remove(UniqueAddress node)
+    {
+        if (_scheduledUnreachable.TryGetValue(node, out var cancelable))
+        {
+            cancelable.Cancel();
+        }
+
+        _scheduledUnreachable = _scheduledUnreachable.Remove(node);
+        _pendingUnreachable = _pendingUnreachable.Remove(node);
+    }
+}