fix(cluster): leader-auto-down restarts a node its peer has shut out; higher address restarts when both are (#132)

A node frozen past failure detection + StableAfter, or both sides of a partition
that heals after each downed the other, used to stay a second active node until an
operator restarted one: the peer quarantines the removed UID and Akka.Cluster
1.5.62 never acts on ThisActorSystemQuarantinedEvent.

ShutOutRestart (child of the LeaderAutoDown downing actor; opt-out via
ScadaBridge:Cluster:LeaderAutoDown:RestartWhenShutOut, default on for
leader-auto-down only) watches two signals:
- S1 ThisActorSystemQuarantinedEvent: the peer quarantined this node. If the peer
  is still a live member here, only this side can be shut out -> restart now.
- S2 AssociationErrorEvent 'has a UID that has been quarantined': the peer still
  runs the incarnation this node removed. A healed partition yields only S2, on
  both sides. Wait ShutOutGrace (60s) for a frozen peer to restart itself, then
  tiebreak.
Tiebreak: the higher address (Akka Member.AddressOrdering) restarts, the lower
stays and restarts only if still split one grace later. Removed peers are probed
(Identify, 5s -> 60s back-off) since nothing else contacts them after mutual
removal. Restart = Cluster.Down(self) -> existing self-down path.

The Host now exits 75 after an unplanned ActorSystem termination (and sets it on
the Windows service before it reports stopped): SCM recovery with failureflag 1
only restarts a service that stops with a non-zero code, and the watchdog used
to exit 0. Shipped default strategy stays auto-down.

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

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

@@ -144,6 +144,61 @@ public class ClusterOptions
     /// <c>docs/plans/2026-11-09-tls-production-profile-design.md</c>.
     /// </summary>
     public ClusterRemotingTlsOptions RemotingTls { get; set; } = new();
+
+    /// <summary>
+    /// Settings that apply only when <see cref="SplitBrainResolverStrategy"/> is
+    /// <c>leader-auto-down</c> (<c>ScadaBridge:Cluster:LeaderAutoDown</c>, ScadaBridge#132).
+    /// Inert under every other strategy: <c>AkkaHostedService.BuildHocon</c> emits none of it.
+    /// </summary>
+    public LeaderAutoDownOptions LeaderAutoDown { get; set; } = new();
+}
+
+/// <summary>
+/// <c>leader-auto-down</c>-only settings (ScadaBridge#132): what a node does when it learns that
+/// its pair peer has <b>shut it out</b> — removed and quarantined it, so it can never talk to that
+/// peer again until it restarts.
+/// </summary>
+/// <remarks>
+/// <para>
+/// <b>Why it exists.</b> A node that was frozen (VM pause, GC stall, <c>docker pause</c>) past the
+/// failure-detection window plus <see cref="ClusterOptions.StableAfter"/> is downed and removed by
+/// its peer. When it resumes, every association is refused with "the remote system has quarantined
+/// this system" — Akka.Remote publishes <c>ThisActorSystemQuarantinedEvent</c>, but Akka.Cluster
+/// 1.5.62 never acts on it, so the frozen node downs the peer in its own one-node view and becomes
+/// a second, permanent active node. The same thing happens to <b>both</b> nodes after a real
+/// network partition heals.
+/// </para>
+/// <para>
+/// <b>What it does</b> (with <see cref="RestartWhenShutOut"/> on): a node that is shut out by its
+/// peer while it still counts that peer as a member restarts itself (it is the stale side). When it
+/// has already removed the peer too — both sides shut each other out — a deterministic tiebreak
+/// both nodes compute identically picks ONE side: the node with the <b>higher</b> address
+/// (host, then port; Akka's own member address ordering) restarts, the lower stays. The lower node
+/// restarts too only if it is still being shut out after <see cref="ShutOutGrace"/>. "Restart" is
+/// the existing self-down path: <c>Cluster.Down(self)</c>, the <c>ActorSystem</c> terminates, the
+/// Host exits non-zero and the service supervisor restarts it as a fresh incarnation that rejoins
+/// as standby. Implemented by <c>ShutOutRestart</c> in the Host.
+/// </para>
+/// </remarks>
+public sealed class LeaderAutoDownOptions
+{
+    /// <summary>
+    /// Whether a node shut out by its pair peer restarts itself (see the class remarks). Default
+    /// <see langword="true"/> — on whenever <c>leader-auto-down</c> is selected. Set
+    /// <see langword="false"/> to get the old behaviour back: a frozen node or a healed partition
+    /// leaves two active nodes until an operator restarts one.
+    /// </summary>
+    public bool RestartWhenShutOut { get; set; } = true;
+
+    /// <summary>
+    /// How long the tiebreak WINNER (the lower-address node) waits for its peer to restart before
+    /// it re-checks. At the end of the grace it restarts only if the peer is still not a member AND
+    /// it was shut out again in the last few seconds — i.e. the peer never restarted, so this node
+    /// is the stale side after all. Must cover the peer's worst-case shut-down time (its old
+    /// incarnation keeps refusing until its <c>ActorSystem</c> has terminated). Default 60s.
+    /// Validated <c>&gt; 0</c> at boot when the feature is on.
+    /// </summary>
+    public TimeSpan ShutOutGrace { get; set; } = TimeSpan.FromSeconds(60);
 }
 
 /// <summary>

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

@@ -74,6 +74,17 @@ public sealed class ClusterOptionsValidator : OptionsValidatorBase<ClusterOption
             $"FailureDetectionThreshold ({options.FailureDetectionThreshold}); otherwise nodes are " +
             "declared unreachable before a heartbeat can arrive.");
 
+        // ScadaBridge#132: the shut-out restart grace is only consulted under leader-auto-down
+        // with the feature on; a zero/negative grace would make the tiebreak winner restart on
+        // the spot, i.e. BOTH nodes of a healed partition restart together.
+        var isLeaderAutoDown = string.Equals(
+            options.SplitBrainResolverStrategy, "leader-auto-down", StringComparison.OrdinalIgnoreCase);
+        var leaderAutoDown = options.LeaderAutoDown ?? new LeaderAutoDownOptions();
+        builder.RequireThat(
+            !isLeaderAutoDown || !leaderAutoDown.RestartWhenShutOut || leaderAutoDown.ShutOutGrace > TimeSpan.Zero,
+            "ClusterOptions.LeaderAutoDown.ShutOutGrace must be a positive duration when "
+            + "LeaderAutoDown.RestartWhenShutOut is on (ScadaBridge#132).");
+
         // 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(

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

@@ -34,8 +34,20 @@ public class AkkaHostedService : IHostedService
     private readonly CommunicationOptions _communicationOptions;
     private readonly ILogger<AkkaHostedService> _logger;
     private readonly IHostApplicationLifetime? _appLifetime;
+    private readonly IHostLifetime? _hostLifetime;
     private ActorSystem? _actorSystem;
 
+    /// <summary>
+    /// Process exit code after an ActorSystem termination outside host shutdown (a self-down,
+    /// or a ScadaBridge#132 shut-out restart): 75, <c>EX_TEMPFAIL</c> — "temporary failure, try
+    /// again". It must be NON-zero: docker <c>restart: unless-stopped</c> restarts on any exit, but
+    /// Windows service recovery (<c>sc.exe failure … restart</c> with <c>failureflag 1</c>, set by
+    /// <c>deploy/wonder-app-vd03/install.ps1</c>) applies its restart actions to a service that
+    /// reports <c>SERVICE_STOPPED</c> only when the reported exit code is not 0. A clean
+    /// <c>StopApplication()</c> alone reports 0, so the node would stay stopped.
+    /// </summary>
+    public const int SelfDownExitCode = 75;
+
     /// <summary>
     /// Flips to <c>true</c> the moment <see cref="StopAsync"/> begins, so the
     /// <see cref="ActorSystem.WhenTerminated"/> watchdog can tell an
@@ -84,14 +96,17 @@ public class AkkaHostedService : IHostedService
     /// <param name="communicationOptions">The communication configuration options.</param>
     /// <param name="logger">The logger instance.</param>
     /// <param name="appLifetime">The host application lifetime; optional so existing construction sites (tests, DI variants) keep working without it.</param>
+    /// <param name="hostLifetime">The host lifetime; when it is the Windows service lifetime, the self-down exit code is reported to the SCM through it. Optional.</param>
     public AkkaHostedService(
         IServiceProvider serviceProvider,
         IOptions<NodeOptions> nodeOptions,
         IOptions<ClusterOptions> clusterOptions,
         IOptions<CommunicationOptions> communicationOptions,
         ILogger<AkkaHostedService> logger,
-        IHostApplicationLifetime? appLifetime = null)
+        IHostApplicationLifetime? appLifetime = null,
+        IHostLifetime? hostLifetime = null)
     {
+        _hostLifetime = hostLifetime;
         _serviceProvider = serviceProvider;
         _nodeOptions = nodeOptions.Value;
         _clusterOptions = clusterOptions.Value;
@@ -212,8 +227,11 @@ public class AkkaHostedService : IHostedService
             {
                 if (_stopRequested) return;
                 _logger.LogCritical(
-                    "ActorSystem terminated outside host shutdown (SBR down / run-coordinated-shutdown-when-down). "
-                    + "Stopping the host process so the service supervisor restarts this node as a fresh incarnation.");
+                    "ActorSystem terminated outside host shutdown (SBR down / run-coordinated-shutdown-when-down / "
+                    + "leader-auto-down shut-out restart). Stopping the host process with exit code {ExitCode} so the "
+                    + "service supervisor restarts this node as a fresh incarnation.", SelfDownExitCode);
+                _terminatedOutsideHostShutdown = true;
+                MarkServiceSelfDownExit(_hostLifetime);
                 _appLifetime?.StopApplication();
             }, TaskContinuationOptions.ExecuteSynchronously);
 
@@ -223,6 +241,37 @@ public class AkkaHostedService : IHostedService
         }
     }
 
+    /// <summary>
+    /// <see langword="true"/> once the <see cref="ActorSystem"/> has terminated outside
+    /// <see cref="StopAsync"/> (a self-down or a shut-out restart). <c>Program</c> reads it after
+    /// <c>RunAsync</c> returns and exits with <see cref="SelfDownExitCode"/>; it is a flag rather than a
+    /// direct write of the process exit code so test hosts that terminate a system never inherit
+    /// a non-zero process exit code.
+    /// </summary>
+    public bool TerminatedOutsideHostShutdown => _terminatedOutsideHostShutdown;
+
+    private volatile bool _terminatedOutsideHostShutdown;
+
+    /// <summary>
+    /// When the host runs as a Windows service, sets <see cref="SelfDownExitCode"/> as the exit code
+    /// the service reports to the SCM when it stops — the value Windows service recovery
+    /// (<c>failureflag 1</c>) keys its restart on. Must run BEFORE <c>StopApplication()</c>: the
+    /// Windows service lifetime reports <c>SERVICE_STOPPED</c> during host shutdown, before
+    /// <c>Program</c> can set the process exit code.
+    /// </summary>
+    /// <param name="hostLifetime">The host lifetime; the Windows service lifetime is a <c>ServiceBase</c>.</param>
+    /// <returns><see langword="true"/> when the exit code was set on a Windows service lifetime.</returns>
+    internal static bool MarkServiceSelfDownExit(IHostLifetime? hostLifetime)
+    {
+        if (OperatingSystem.IsWindows() && hostLifetime is System.ServiceProcess.ServiceBase service)
+        {
+            service.ExitCode = SelfDownExitCode;
+            return true;
+        }
+
+        return false;
+    }
+
     /// <summary>
     /// Builds the Akka HOCON configuration document. Every interpolated value is
     /// routed through <see cref="QuoteHocon"/> (string values) so a hostname,
@@ -313,8 +362,12 @@ 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.
+        // ScadaBridge#132: the shut-out restart switch and grace ride in the same block.
+        var shutOut = clusterOptions.LeaderAutoDown ?? new LeaderAutoDownOptions();
         var leaderAutoDownBlock = string.Equals(strategy, "leader-auto-down", StringComparison.OrdinalIgnoreCase)
             ? $"{LeaderAutoDowningProvider.UnreachableAfterPath} = {DurationHocon(clusterOptions.StableAfter)}\n"
+              + $"{ShutOutRestartSettings.EnabledPath} = {(shutOut.RestartWhenShutOut ? "on" : "off")}\n"
+              + $"{ShutOutRestartSettings.GracePath} = {DurationHocon(shutOut.ShutOutGrace)}\n"
             : string.Empty;
 
         // Akka.Remote TLS (ScadaBridge#64, opt-in). Emitted inside dot-netty.tcp only when

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

@@ -88,7 +88,8 @@ public sealed class LeaderAutoDowningProvider : IDowningProvider
                     $"LeaderAutoDowningProvider selected but '{UnreachableAfterPath}' not set");
             }
 
-            return LeaderAutoDown.Props(after.Value, _cluster);
+            return LeaderAutoDown.Props(
+                after.Value, _cluster, ShutOutRestartSettings.FromConfig(_system.Settings.Config));
         }
     }
 
@@ -118,21 +119,29 @@ public sealed class LeaderAutoDowningProvider : IDowningProvider
 /// </summary>
 internal sealed class LeaderAutoDown : LeaderAutoDownBase
 {
+    /// <summary>Child name of the <see cref="ShutOutRestart"/> actor.</summary>
+    internal const string ShutOutRestartName = "shut-out-restart";
+
     private readonly Akka.Cluster.Cluster _cluster;
+    private readonly ShutOutRestartSettings? _shutOut;
 
     /// <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));
+    /// <param name="shutOut">Shut-out restart settings (ScadaBridge#132); null or disabled = no shut-out handling.</param>
+    /// <returns>The props.</returns>
+    public static Props Props(TimeSpan unreachableAfter, Akka.Cluster.Cluster cluster, ShutOutRestartSettings? shutOut = null) =>
+        Akka.Actor.Props.Create(() => new LeaderAutoDown(unreachableAfter, cluster, shutOut));
 
     /// <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)
+    /// <param name="shutOut">Shut-out restart settings (ScadaBridge#132); null or disabled = no shut-out handling.</param>
+    public LeaderAutoDown(TimeSpan unreachableAfter, Akka.Cluster.Cluster cluster, ShutOutRestartSettings? shutOut)
         : base(unreachableAfter)
     {
         _cluster = cluster;
+        _shutOut = shutOut;
     }
 
     /// <inheritdoc />
@@ -145,6 +154,22 @@ internal sealed class LeaderAutoDown : LeaderAutoDownBase
     protected override void PreStart()
     {
         _cluster.Subscribe(Self, typeof(ClusterEvent.IClusterDomainEvent));
+
+        // ScadaBridge#132: the shut-out restart is a separate child so the downing logic above
+        // stays a line-for-line port of Akka's AutoDownBase. A child, not a sibling, so it lives
+        // and dies with this provider's actor and exists only under leader-auto-down.
+        if (_shutOut is { Enabled: true })
+        {
+            Context.ActorOf(ShutOutRestart.Props(_shutOut, _cluster), ShutOutRestartName);
+        }
+        else
+        {
+            Context.GetLogger().Info(
+                "[leader-auto-down] shut-out restart is OFF for {0} (ScadaBridge:Cluster:LeaderAutoDown:RestartWhenShutOut = false): "
+                + "a node its peer has removed and quarantined keeps running as a second active node until it is restarted",
+                _cluster.SelfAddress);
+        }
+
         base.PreStart();
     }
 

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

@@ -0,0 +1,674 @@
+using Akka.Actor;
+using Akka.Cluster;
+using Akka.Configuration;
+using Akka.Event;
+using Akka.Remote;
+
+namespace ZB.MOM.WW.ScadaBridge.Host.Actors;
+
+/// <summary>
+/// Settings for <see cref="ShutOutRestart"/>, read from the <c>scadabridge.cluster.leader-auto-down</c>
+/// HOCON block that <see cref="AkkaHostedService.BuildHocon"/> emits from
+/// <c>ScadaBridge:Cluster:LeaderAutoDown</c> (ScadaBridge#132).
+/// </summary>
+/// <param name="Enabled">Whether a node that is shut out by its peer restarts itself.</param>
+/// <param name="Grace">How long a peer gets to restart itself before this node acts (see <see cref="ShutOutRestart"/>).</param>
+/// <param name="ProbeInterval">First interval between probes of a removed peer, and the fixed interval while a grace runs.</param>
+/// <param name="ProbeMaxInterval">Cap for the probe back-off while no grace runs.</param>
+internal sealed record ShutOutRestartSettings(
+    bool Enabled, TimeSpan Grace, TimeSpan ProbeInterval, TimeSpan ProbeMaxInterval)
+{
+    /// <summary>HOCON switch: <c>on</c>/<c>off</c>. Missing = on.</summary>
+    public const string EnabledPath = "scadabridge.cluster.leader-auto-down.restart-when-shut-out";
+
+    /// <summary>HOCON duration for <see cref="Grace"/>. Missing = 60s.</summary>
+    public const string GracePath = "scadabridge.cluster.leader-auto-down.shut-out-grace";
+
+    /// <summary>HOCON duration for <see cref="ProbeInterval"/>. Missing = 5s. Not bound from appsettings (tests only).</summary>
+    public const string ProbeIntervalPath = "scadabridge.cluster.leader-auto-down.shut-out-probe-interval";
+
+    /// <summary>HOCON duration for <see cref="ProbeMaxInterval"/>. Missing = 60s. Not bound from appsettings (tests only).</summary>
+    public const string ProbeMaxIntervalPath = "scadabridge.cluster.leader-auto-down.shut-out-probe-max-interval";
+
+    /// <summary>
+    /// How recent the last shut-out evidence must be, when a grace ends, for the split to count as
+    /// still in place: three probe intervals, so a missed probe or two does not flip the verdict.
+    /// </summary>
+    public TimeSpan RecentWindow => ProbeInterval * 3;
+
+    /// <summary>Reads the settings, defaulting every missing key.</summary>
+    /// <param name="config">The actor system's configuration.</param>
+    /// <returns>The settings.</returns>
+    public static ShutOutRestartSettings FromConfig(Config config)
+    {
+        var enabled = !config.HasPath(EnabledPath) || config.GetBoolean(EnabledPath, true);
+        return new ShutOutRestartSettings(
+            enabled,
+            Duration(config, GracePath, TimeSpan.FromSeconds(60)),
+            Duration(config, ProbeIntervalPath, TimeSpan.FromSeconds(5)),
+            Duration(config, ProbeMaxIntervalPath, TimeSpan.FromSeconds(60)));
+    }
+
+    private static TimeSpan Duration(Config config, string path, TimeSpan fallback) =>
+        config.HasPath(path) ? config.GetTimeSpan(path, fallback, true) : fallback;
+}
+
+/// <summary>
+/// ScadaBridge#132: makes a node that is split from its pair peer — the peer has downed and removed
+/// it, or each has removed the other — restart itself, so a frozen node or a healed partition does
+/// not leave two active nodes until an operator acts. Runs as a child of the
+/// <see cref="LeaderAutoDown"/> downing actor, so it exists only under <c>leader-auto-down</c>, and
+/// only when <c>ScadaBridge:Cluster:LeaderAutoDown:RestartWhenShutOut</c> is on (the default for
+/// that strategy).
+/// <para>
+/// <b>The signals</b> (Akka.NET 1.5.62, read from the decompiled assemblies and confirmed by the
+/// in-process scenario tests). When a node removes a downed member, its cluster quarantines that
+/// member's UID. After that, each association between the two incarnations fails, and which side
+/// learns what depends on who still counts whom as a member:
+/// <list type="bullet">
+/// <item><b>S1 — "the peer has quarantined THIS node".</b> Akka.Remote publishes
+/// <see cref="ThisActorSystemQuarantinedEvent"/> (<c>EndpointManager</c>, on an
+/// <c>InvalidAssociation</c> whose disassociate reason is <c>Quarantined</c>). Seen by a node that
+/// still talks to a peer which has removed it: the frozen node after <c>docker pause</c>. Akka.Cluster
+/// 1.5.62 never subscribes to it — <c>ClusterCoreDaemon</c> handles only the opposite
+/// <c>QuarantinedEvent</c> — which is the root cause of #132.</item>
+/// <item><b>S2 — "the peer is still running the incarnation THIS node removed".</b> An outbound
+/// handshake that meets a UID this node has quarantined is aborted (<c>ProtocolStateActor</c>,
+/// <c>ForbiddenUidReason</c>: "The remote system has a UID that has been quarantined"), surfacing as
+/// an <see cref="AssociationErrorEvent"/>. Seen by both sides of a partition that healed after each
+/// side downed the other — neither of them sees S1, because each aborts the other's handshake first —
+/// and by the survivor of a freeze when it probes the frozen node.</item>
+/// </list>
+/// </para>
+/// <para>
+/// <b>The decision.</b>
+/// <list type="number">
+/// <item><b>S1 while the peer is still a live member in this node's view</b> (Joining, WeaklyUp, Up
+/// or Leaving): this node has not removed the peer, so it has not quarantined it, so the peer
+/// cannot be shut out by this node — only this side is. This is the frozen node. <b>Restart
+/// now</b>; no tiebreak is needed.</item>
+/// <item><b>S1 after this node has removed the peer too</b>: both sides are shut out.
+/// <b>Tiebreak now</b> (below).</item>
+/// <item><b>S2</b>: the peer's old incarnation is alive. It may be a frozen node about to restart
+/// itself on S1 — so this node first gives it <see cref="ShutOutRestartSettings.Grace"/>. If, at
+/// the end, the peer has neither rejoined nor gone quiet, both sides are shut out: <b>tiebreak</b>.
+/// </item>
+/// <item><b>Tiebreak</b>: the node with the <b>higher</b> address restarts now. The lower one stays
+/// and gives the peer one more <see cref="ShutOutRestartSettings.Grace"/>; it restarts only if the
+/// peer has still not rejoined and is still producing evidence at the end — the single-sided case
+/// where the stale side happened to be the lower address.</item>
+/// </list>
+/// "Quiet" means no S1/S2 from that peer for <see cref="ShutOutRestartSettings.RecentWindow"/>: the
+/// old incarnation is gone (restarting, or dead), so there is nothing left to split from.
+/// </para>
+/// <para>
+/// <b>Why the address tiebreak, not member age.</b> Both nodes must reach the same answer without
+/// talking to each other, from state they already hold. The two addresses are in every signal and
+/// in both nodes' seed lists, so the comparison is identical on both sides by construction. Age
+/// (up-number) is not: by the time a signal fires both sides have removed each other, so the peer's
+/// up-number survives only as remembered history, which differs if a join never converged before
+/// the split or either side has restarted since. A disagreement there means both restart or neither
+/// does — exactly what the tiebreak exists to prevent. The ordering is Akka's own
+/// <see cref="Member.AddressOrdering"/> (host, then port), which also picks the Akka leader of the
+/// re-formed pair, and matches the bootstrap guard, where the lower address is the preferred founder.
+/// </para>
+/// <para>
+/// <b>Probing.</b> After both sides have removed each other neither sends the other anything, so a
+/// healed partition would stay silent and neither signal would ever fire. This actor therefore
+/// sends an <see cref="Identify"/> to every peer it saw removed after being downed, with a back-off
+/// from <see cref="ShutOutRestartSettings.ProbeInterval"/> to
+/// <see cref="ShutOutRestartSettings.ProbeMaxInterval"/> (fixed at the short interval while a grace
+/// runs), until that address is a member again. A probe of a restarted incarnation (new UID) is
+/// harmless — the new incarnation is joining anyway; a probe of the old incarnation produces S2
+/// here and S1 or S2 there.
+/// </para>
+/// <para>
+/// <b>The restart</b> reuses the existing self-down path: <c>Cluster.Down(self)</c> ⇒
+/// <c>run-coordinated-shutdown-when-down</c> ⇒ the <c>ActorSystem</c> terminates ⇒ the Host's
+/// <c>WhenTerminated</c> watchdog exits the process with <see cref="AkkaHostedService.SelfDownExitCode"/>
+/// ⇒ the service supervisor (docker <c>restart: unless-stopped</c>, Windows service recovery with
+/// <c>failureflag 1</c>) starts a fresh incarnation that rejoins its peer as standby. If the system
+/// has not terminated <see cref="FallbackShutdownAfter"/> after the Down, <c>CoordinatedShutdown</c>
+/// is run directly.
+/// </para>
+/// </summary>
+internal sealed class ShutOutRestart : ShutOutRestartBase
+{
+    /// <summary>How long after <c>Cluster.Down(self)</c> the direct <c>CoordinatedShutdown</c> fallback runs.</summary>
+    internal static readonly TimeSpan FallbackShutdownAfter = TimeSpan.FromSeconds(20);
+
+    private readonly Akka.Cluster.Cluster _cluster;
+    private ICancelable? _fallback;
+
+    /// <summary>Props for the actor.</summary>
+    /// <param name="settings">The settings.</param>
+    /// <param name="cluster">The cluster extension.</param>
+    /// <returns>The props.</returns>
+    public static Props Props(ShutOutRestartSettings settings, Akka.Cluster.Cluster cluster) =>
+        Akka.Actor.Props.Create(() => new ShutOutRestart(settings, cluster));
+
+    /// <summary>Creates the actor.</summary>
+    /// <param name="settings">The settings.</param>
+    /// <param name="cluster">The cluster extension.</param>
+    public ShutOutRestart(ShutOutRestartSettings settings, Akka.Cluster.Cluster cluster)
+        : base(settings)
+    {
+        _cluster = cluster;
+    }
+
+    /// <inheritdoc />
+    public override Address SelfAddress => _cluster.SelfAddress;
+
+    /// <inheritdoc />
+    public override IScheduler Scheduler => Context.System.Scheduler;
+
+    /// <inheritdoc />
+    protected override IEnumerable<Address> SeedNodes => _cluster.Settings.SeedNodes;
+
+    /// <inheritdoc />
+    protected override void PreStart()
+    {
+        _cluster.Subscribe(Self, ClusterEvent.SubscriptionInitialStateMode.InitialStateAsSnapshot,
+            typeof(ClusterEvent.IMemberEvent));
+        Context.System.EventStream.Subscribe(Self, typeof(ThisActorSystemQuarantinedEvent));
+        Context.System.EventStream.Subscribe(Self, typeof(AssociationErrorEvent));
+        base.PreStart();
+    }
+
+    /// <inheritdoc />
+    protected override void PostStop()
+    {
+        _cluster.Unsubscribe(Self);
+        Context.System.EventStream.Unsubscribe(Self);
+        _fallback?.Cancel();
+        base.PostStop();
+    }
+
+    /// <inheritdoc />
+    protected override void RestartSelf()
+    {
+        _cluster.Down(_cluster.SelfAddress);
+        _fallback = Scheduler.ScheduleTellOnceCancelable(FallbackShutdownAfter, Self, FallbackShutdown.Instance, Self);
+    }
+
+    /// <inheritdoc />
+    protected override void Probe(Address peer)
+    {
+        Context.ActorSelection(new RootActorPath(peer) / "system" / "cluster" / "core" / "daemon")
+            .Tell(new Identify(ProbeMessageId), Self);
+    }
+
+    /// <inheritdoc />
+    protected override void OnFallbackShutdown()
+    {
+        if (!Context.System.WhenTerminated.IsCompleted)
+        {
+            Log.Warning(
+                "[leader-auto-down] {0} was downed for a shut-out restart {1} ago but its actor system is still running; running CoordinatedShutdown directly",
+                SelfAddress, FallbackShutdownAfter);
+            CoordinatedShutdown.Get(Context.System).Run(CoordinatedShutdown.ClusterDowningReason.Instance);
+        }
+    }
+}
+
+/// <summary>
+/// The cluster-free decision logic of <see cref="ShutOutRestart"/>, split out (as
+/// <see cref="LeaderAutoDownBase"/> is) so the tiebreak and the single- and both-sides cases can be
+/// unit tested by feeding events straight to the actor. Every decision is logged with the
+/// <c>[leader-auto-down]</c> prefix; the restart / stay verdicts at Warning.
+/// </summary>
+internal abstract class ShutOutRestartBase : UntypedActor
+{
+    /// <summary>The <see cref="Identify.MessageId"/> carried by every probe.</summary>
+    internal const string ProbeMessageId = "scadabridge-shut-out-probe";
+
+    /// <summary>Timer: probe <see cref="Peer"/> again.</summary>
+    /// <param name="Peer">The removed peer to probe.</param>
+    internal sealed record ProbeTick(Address Peer);
+
+    /// <summary>Timer: the grace for <see cref="Peer"/> has ended.</summary>
+    /// <param name="Peer">The peer this node is split from.</param>
+    internal sealed record GraceExpired(Address Peer);
+
+    /// <summary>Timer: the direct-shutdown fallback after a restart was requested.</summary>
+    internal sealed class FallbackShutdown
+    {
+        /// <summary>The singleton instance.</summary>
+        public static readonly FallbackShutdown Instance = new();
+
+        private FallbackShutdown()
+        {
+        }
+    }
+
+    /// <summary>A peer identity independent of the transport scheme (the throttler adapter rewrites it).</summary>
+    private readonly record struct PeerKey(string System, string? Host, int? Port)
+    {
+        public static PeerKey Of(Address address) => new(address.System, address.Host, address.Port);
+    }
+
+    private sealed class ProbeState
+    {
+        public required Address Address { get; init; }
+        public TimeSpan NextInterval { get; set; }
+        public ICancelable? Timer { get; set; }
+    }
+
+    /// <summary>Phase of a detected split with one peer.</summary>
+    private enum SplitPhase
+    {
+        /// <summary>S2 seen: the peer's old incarnation is alive; it gets a grace to restart itself.</summary>
+        Suspect,
+
+        /// <summary>Tiebreak said stay: the peer is expected to restart; it gets a grace to do so.</summary>
+        Winner,
+    }
+
+    private sealed class SplitState
+    {
+        public required SplitPhase Phase { get; set; }
+        public required ICancelable Timer { get; set; }
+        public DateTime LastEvidenceUtc { get; set; }
+    }
+
+    private readonly ShutOutRestartSettings _settings;
+    private readonly Dictionary<PeerKey, Member> _members = new();
+    private readonly HashSet<PeerKey> _knownPeers = new();
+    private readonly HashSet<PeerKey> _ignoredLogged = new();
+    private readonly Dictionary<PeerKey, ProbeState> _probing = new();
+    private readonly Dictionary<PeerKey, SplitState> _splits = new();
+    private bool _restarting;
+
+    /// <summary>Creates the base logic.</summary>
+    /// <param name="settings">The settings.</param>
+    protected ShutOutRestartBase(ShutOutRestartSettings settings)
+    {
+        _settings = settings;
+    }
+
+    /// <summary>Decision log.</summary>
+    protected ILoggingAdapter Log { get; } = Context.GetLogger();
+
+    /// <summary>This node's address.</summary>
+    public abstract Address SelfAddress { get; }
+
+    /// <summary>The scheduler used for the probe and grace timers.</summary>
+    public abstract IScheduler Scheduler { get; }
+
+    /// <summary>The configured seed nodes; peers of this pair even before they were ever seen as members.</summary>
+    protected abstract IEnumerable<Address> SeedNodes { get; }
+
+    /// <summary>The clock; overridable for tests.</summary>
+    protected virtual DateTime UtcNow => DateTime.UtcNow;
+
+    /// <summary>Restarts this node (called at most once).</summary>
+    protected abstract void RestartSelf();
+
+    /// <summary>Sends one probe to <paramref name="peer"/>.</summary>
+    /// <param name="peer">The removed peer.</param>
+    protected abstract void Probe(Address peer);
+
+    /// <summary>Runs the direct-shutdown fallback, if the restart has not taken effect.</summary>
+    protected virtual void OnFallbackShutdown()
+    {
+    }
+
+    /// <summary>
+    /// Whether an association failure is S2: the handshake met a UID this node has quarantined.
+    /// Akka.Remote 1.5.62 carries no typed reason outward — the protocol layer's
+    /// <c>ForbiddenUidReason</c> becomes "The remote system has a UID that has been quarantined.
+    /// Association aborted." and the endpoint layer's <c>HopelessAssociation</c> says "The remote
+    /// system [..] has a UID [..] that has been quarantined." — so the cause chain is matched on that
+    /// wording. Pinned by the in-process scenario tests, which fail if an Akka upgrade rewords it.
+    /// </summary>
+    /// <param name="cause">The association error's cause.</param>
+    /// <returns><see langword="true"/> when the failure is a refused, quarantined UID.</returns>
+    internal static bool IsQuarantinedUidRefusal(Exception? cause)
+    {
+        for (var depth = 0; cause is not null && depth < 8; depth++)
+        {
+            var message = cause.Message;
+            if (message.Contains("has a UID", StringComparison.Ordinal)
+                && message.Contains("that has been quarantined", StringComparison.Ordinal))
+            {
+                return true;
+            }
+
+            if (cause is AggregateException aggregate && aggregate.InnerExceptions.Any(IsQuarantinedUidRefusal))
+            {
+                return true;
+            }
+
+            cause = cause.InnerException;
+        }
+
+        return false;
+    }
+
+    /// <inheritdoc />
+    protected override void PreStart()
+    {
+        foreach (var seed in SeedNodes)
+        {
+            AddKnownPeer(seed);
+        }
+
+        Log.Info(
+            "[leader-auto-down] shut-out restart is ON for {0}: a node split from its pair peer restarts itself — at once if only it "
+            + "was removed, otherwise the higher address restarts after a {1} grace (ScadaBridge#132)",
+            SelfAddress, _settings.Grace);
+    }
+
+    /// <inheritdoc />
+    protected override void PostStop()
+    {
+        CancelAllTimers();
+    }
+
+    /// <inheritdoc />
+    protected override void OnReceive(object message)
+    {
+        switch (message)
+        {
+            case ClusterEvent.CurrentClusterState state:
+                _members.Clear();
+                foreach (var member in state.Members)
+                {
+                    TrackMember(member);
+                }
+                break;
+
+            case ClusterEvent.MemberRemoved removed:
+                _members.Remove(PeerKey.Of(removed.Member.Address));
+                if (removed.PreviousStatus == MemberStatus.Down && !IsSelf(removed.Member.Address))
+                {
+                    StartProbing(removed.Member.Address);
+                }
+                break;
+
+            case ClusterEvent.IMemberEvent memberEvent:
+                TrackMember(memberEvent.Member);
+                break;
+
+            case ThisActorSystemQuarantinedEvent shutOut:
+                OnShutOutByPeer(shutOut.RemoteAddress);
+                break;
+
+            case AssociationErrorEvent error:
+                if (IsQuarantinedUidRefusal(error.Cause))
+                {
+                    OnPeerStillRunningRemovedIncarnation(error.RemoteAddress);
+                }
+                break;
+
+            case ProbeTick tick:
+                OnProbeTick(tick.Peer);
+                break;
+
+            case GraceExpired expired:
+                OnGraceExpired(expired.Peer);
+                break;
+
+            case FallbackShutdown:
+                OnFallbackShutdown();
+                break;
+
+            case ActorIdentity:
+                // A probe answered by a reachable incarnation — nothing to do.
+                break;
+
+            default:
+                Unhandled(message);
+                break;
+        }
+    }
+
+    private static bool IsLiveStatus(MemberStatus status) =>
+        status is MemberStatus.Joining or MemberStatus.WeaklyUp or MemberStatus.Up or MemberStatus.Leaving;
+
+    private bool IsSelf(Address address) => PeerKey.Of(address) == PeerKey.Of(SelfAddress);
+
+    private void AddKnownPeer(Address address)
+    {
+        if (!IsSelf(address))
+        {
+            _knownPeers.Add(PeerKey.Of(address));
+        }
+    }
+
+    private void TrackMember(Member member)
+    {
+        var key = PeerKey.Of(member.Address);
+        _members[key] = member;
+        AddKnownPeer(member.Address);
+        if (IsLiveStatus(member.Status) && _probing.Remove(key, out var probe))
+        {
+            probe.Timer?.Cancel();
+        }
+    }
+
+    private bool IsLiveMember(Address address, out MemberStatus status)
+    {
+        if (_members.TryGetValue(PeerKey.Of(address), out var member) && IsLiveStatus(member.Status))
+        {
+            status = member.Status;
+            return true;
+        }
+
+        status = MemberStatus.Removed;
+        return false;
+    }
+
+    /// <summary>Common gate for both signals: ignore after a restart, for self, and for strangers.</summary>
+    private bool IsRelevantPeer(Address peer, string what)
+    {
+        if (_restarting || IsSelf(peer))
+        {
+            return false;
+        }
+
+        var key = PeerKey.Of(peer);
+        if (_knownPeers.Contains(key))
+        {
+            return true;
+        }
+
+        if (_ignoredLogged.Add(key))
+        {
+            Log.Info("[leader-auto-down] {0} {1}, but it is not a seed or member of this cluster; ignored", peer, what);
+        }
+
+        return false;
+    }
+
+    /// <summary>S1: <paramref name="peer"/> has quarantined this node.</summary>
+    private void OnShutOutByPeer(Address peer)
+    {
+        if (!IsRelevantPeer(peer, $"has quarantined this node ({SelfAddress})"))
+        {
+            return;
+        }
+
+        if (IsLiveMember(peer, out var status))
+        {
+            Log.Warning(
+                "[leader-auto-down] shut out by {0}: it has removed and quarantined this node ({1}) while this node still counts it as a member ({2}), "
+                + "so only this side is shut out; tiebreak not needed — this node RESTARTS to rejoin as a fresh incarnation",
+                peer, SelfAddress, status);
+            Restart();
+            return;
+        }
+
+        if (_splits.TryGetValue(PeerKey.Of(peer), out var split))
+        {
+            split.LastEvidenceUtc = UtcNow;
+            if (split.Phase == SplitPhase.Winner)
+            {
+                return;
+            }
+
+            // Suspect → the peer has quarantined us too: both sides are shut out, decide now.
+            split.Timer.Cancel();
+            _splits.Remove(PeerKey.Of(peer));
+        }
+
+        Tiebreak(peer, $"shut out by {peer}: it has quarantined this node ({SelfAddress}), and this node has removed it too");
+    }
+
+    /// <summary>S2: <paramref name="peer"/> is still running the incarnation this node removed.</summary>
+    private void OnPeerStillRunningRemovedIncarnation(Address peer)
+    {
+        if (!IsRelevantPeer(peer, "refused a handshake with a UID this node has quarantined"))
+        {
+            return;
+        }
+
+        var key = PeerKey.Of(peer);
+        if (_splits.TryGetValue(key, out var split))
+        {
+            split.LastEvidenceUtc = UtcNow;
+            return;
+        }
+
+        if (IsLiveMember(peer, out _))
+        {
+            // A live member with a quarantined UID is a transient of Akka's own quarantine path
+            // (it is about to be marked terminated and downed); the downing provider handles it.
+            return;
+        }
+
+        Log.Warning(
+            "[leader-auto-down] split from {0}: it is still running the incarnation this node ({1}) removed, so this node has shut it out. "
+            + "Giving it {2} to restart itself before the tiebreak (a frozen node that was removed restarts on its own)",
+            peer, SelfAddress, _settings.Grace);
+        StartSplit(peer, SplitPhase.Suspect);
+    }
+
+    private void Tiebreak(Address peer, string why)
+    {
+        if (Member.AddressOrdering.Compare(SelfAddress, peer) > 0)
+        {
+            Log.Warning(
+                "[leader-auto-down] {0}; both sides are shut out. Tiebreak says RESTART: this node ({1}) has the higher address",
+                why, SelfAddress);
+            Restart();
+            return;
+        }
+
+        Log.Warning(
+            "[leader-auto-down] {0}; both sides are shut out. Tiebreak says STAY: this node ({1}) has the lower address, so {2} restarts. "
+            + "This node restarts too only if {2} has not rejoined and is still split from it after {3}",
+            why, SelfAddress, peer, _settings.Grace);
+        StartSplit(peer, SplitPhase.Winner);
+    }
+
+    private void StartSplit(Address peer, SplitPhase phase)
+    {
+        _splits[PeerKey.Of(peer)] = new SplitState
+        {
+            Phase = phase,
+            Timer = Scheduler.ScheduleTellOnceCancelable(_settings.Grace, Self, new GraceExpired(peer), Self),
+            LastEvidenceUtc = UtcNow,
+        };
+
+        // Probe at the short interval during the grace, so "still split" is measured.
+        StartProbing(peer, restartBackOff: true);
+    }
+
+    private void OnGraceExpired(Address peer)
+    {
+        if (_restarting || !_splits.Remove(PeerKey.Of(peer), out var split))
+        {
+            return;
+        }
+
+        if (IsLiveMember(peer, out var status))
+        {
+            Log.Info(
+                "[leader-auto-down] {0} rejoined ({1}) within the shut-out grace; this node ({2}) stays",
+                peer, status, SelfAddress);
+            return;
+        }
+
+        var sinceLast = UtcNow - split.LastEvidenceUtc;
+        if (sinceLast > _settings.RecentWindow)
+        {
+            Log.Info(
+                "[leader-auto-down] {0} has not rejoined yet, but nothing of its old incarnation has been seen for {1:0}s (it is restarting or down); "
+                + "this node ({2}) stays",
+                peer, sinceLast.TotalSeconds, SelfAddress);
+            return;
+        }
+
+        if (split.Phase == SplitPhase.Suspect)
+        {
+            Tiebreak(peer,
+                $"split from {peer}: its old incarnation is still running {_settings.Grace} later and it has not restarted itself");
+            return;
+        }
+
+        Log.Warning(
+            "[leader-auto-down] {0} is still split from this node ({1}) {2} after the tiebreak (last seen {3:0}s ago) and has not rejoined, "
+            + "so this node is the stale side after all; this node RESTARTS",
+            peer, SelfAddress, _settings.Grace, sinceLast.TotalSeconds);
+        Restart();
+    }
+
+    private void StartProbing(Address peer, bool restartBackOff = false)
+    {
+        var key = PeerKey.Of(peer);
+        if (_probing.TryGetValue(key, out var existing))
+        {
+            if (!restartBackOff)
+            {
+                return;
+            }
+
+            existing.Timer?.Cancel();
+        }
+
+        var state = existing ?? new ProbeState { Address = peer };
+        state.NextInterval = _settings.ProbeInterval;
+        state.Timer = Scheduler.ScheduleTellOnceCancelable(state.NextInterval, Self, new ProbeTick(peer), Self);
+        _probing[key] = state;
+    }
+
+    private void OnProbeTick(Address peer)
+    {
+        var key = PeerKey.Of(peer);
+        if (_restarting || !_probing.TryGetValue(key, out var state))
+        {
+            return;
+        }
+
+        Probe(state.Address);
+
+        // Fixed short interval while a grace runs (to measure "still split"); otherwise back off.
+        state.NextInterval = _splits.ContainsKey(key)
+            ? _settings.ProbeInterval
+            : TimeSpan.FromTicks(Math.Min(state.NextInterval.Ticks * 2, _settings.ProbeMaxInterval.Ticks));
+        state.Timer = Scheduler.ScheduleTellOnceCancelable(state.NextInterval, Self, new ProbeTick(peer), Self);
+    }
+
+    private void Restart()
+    {
+        _restarting = true;
+        CancelAllTimers();
+        _probing.Clear();
+        _splits.Clear();
+        RestartSelf();
+    }
+
+    private void CancelAllTimers()
+    {
+        foreach (var probe in _probing.Values)
+        {
+            probe.Timer?.Cancel();
+        }
+
+        foreach (var split in _splits.Values)
+        {
+            split.Timer.Cancel();
+        }
+    }
+}

src/ZB.MOM.WW.ScadaBridge.Host/Program.cs

@@ -157,6 +157,11 @@ Log.Logger = ZB.MOM.WW.ScadaBridge.Host.ScadaBridgeLogging
 // The catch below uses it to tell that routine stop from a genuine fatal exit (ScadaBridge#76).
 var hostStopping = CancellationToken.None;
 
+// The Akka hosted service, captured with hostStopping (RunAsync disposes the host's service
+// provider), so the finally block can turn an Akka watchdog stop (self-down / ScadaBridge#132
+// shut-out restart) into a non-zero exit code whichever way RunAsync unwound.
+AkkaHostedService? akkaHostedService = null;
+
 try
 {
     Log.Information("Starting ScadaBridge host as {Role} on {Hostname}", nodeRole, nodeHostname);
@@ -670,6 +675,7 @@ try
 
         var app = builder.Build();
         hostStopping = app.Lifetime.ApplicationStopping;
+        akkaHostedService = app.Services.GetService<AkkaHostedService>();
 
         // Apply or validate database migrations (skip when running in test harness)
         if (!string.Equals(configuration["ScadaBridge:Database:SkipMigrations"], "true", StringComparison.OrdinalIgnoreCase))
@@ -864,6 +870,7 @@ try
 
         var app = builder.Build();
         hostStopping = app.Lifetime.ApplicationStopping;
+        akkaHostedService = app.Services.GetService<AkkaHostedService>();
 
         // Endpoint routing middleware. The gRPC service mapping below and the
         // /metrics scrape endpoint both run on endpoint routing, so UseRouting()
@@ -934,6 +941,8 @@ catch (Exception ex)
 }
 finally
 {
+    Program.SetSelfDownExitCode(akkaHostedService);
+
     await Log.CloseAndFlushAsync();
 }
 
@@ -955,6 +964,25 @@ public partial class Program
     internal static bool IsGracefulShutdown(Exception ex, CancellationToken hostStopping) =>
         ex is OperationCanceledException && hostStopping.IsCancellationRequested;
 
+    /// <summary>
+    /// After <c>RunAsync</c> returns: when the stop was the Akka watchdog's (the ActorSystem
+    /// terminated outside host shutdown — a self-down or a ScadaBridge#132 shut-out restart), exit
+    /// with the non-zero <see cref="AkkaHostedService.SelfDownExitCode"/> so the service supervisor
+    /// restarts the node. Docker <c>restart: unless-stopped</c> restarts on any exit; Windows service
+    /// recovery gets the code through the service lifetime (<c>AkkaHostedService.MarkServiceSelfDownExit</c>).
+    /// </summary>
+    /// <param name="akka">The host's Akka hosted service; null if the host never built.</param>
+    internal static void SetSelfDownExitCode(AkkaHostedService? akka)
+    {
+        if (akka is { TerminatedOutsideHostShutdown: true })
+        {
+            Environment.ExitCode = AkkaHostedService.SelfDownExitCode;
+            Log.Warning(
+                "ScadaBridge host stopped because its actor system terminated (self-down / shut-out restart); exiting with code {ExitCode} so the service supervisor restarts it",
+                AkkaHostedService.SelfDownExitCode);
+        }
+    }
+
     /// <summary>
     /// Extracts the distinct TCP ports from an ASP.NET Core server-URLs string (the
     /// <c>ASPNETCORE_URLS</c> / <c>--urls</c> value). Port-only view of