Backport MNTR conductor reliability to v1.5 (#8431 + #8485) (#8488)

Barrier failures now actually fail the barrier: three Player.cs sites replied with the FSM-inherited Failure type, so the barrier ask completed successfully and the failure was silently swallowed - the asking node walked on unsynchronized and the breakage surfaced elsewhere as an unrelated-looking flake. All three now reply Status.Failure, plus the discarded-ask cleanup and the runner mislabeling fix.

Conductor keeps re-registered nodes: a restarting node's stale ClientDisconnected was matched by role name and evicted the fresh registration, then the barrier coordinator dropped the evicted node's arrivals with no reply, hanging it for the full ask timeout. Disconnects are now matched against the registered FSM identity, unregistered arrivals get an explicit BarrierResult(false), and ReleaseAll detaches the shared event-loop groups.

Validated locally with revert-proven tests and three consecutive green ReDeployment MNTR runs. With barrier failures now honest, latent v1.5 spec failures may surface attributed to the node that actually broke - the intended effect. Dev's node-hang kill backstop was deliberately not taken: it needs Process.Kill(entireProcessTree:), unavailable on netstandard2.0; needs a netstandard-safe follow-up.

Author: Aaronontheweb

src/core/Akka.MultiNode.TestAdapter/Internal/MultiNodeTestRunner.cs

@@ -298,7 +298,7 @@ void OutputHandler(object sender, DataReceivedEventArgs eventArgs)
                     }
                     else
                     {
-                        _sinkCoordinator.Tell(new NodeCompletedSpecWithFail(_test.Node, _test.Role, _test.DisplayName + " passed."));
+                        _sinkCoordinator.Tell(new NodeCompletedSpecWithFail(_test.Node, _test.Role, _test.DisplayName + " failed."));
                     }
                 };
                 opt.OutputDataReceived = OutputHandler;

src/core/Akka.Remote.TestKit.Tests/BarrierSpec.cs

@@ -343,6 +343,28 @@ public void A_BarrierCoordinator_must_fail_if_a_node_registers_twice()
                 , nodeB), msg.Exception);
         }
 
+        [Fact(DisplayName = "BarrierCoordinator should fail an arrival from an unregistered client while waiting")]
+        public async Task A_BarrierCoordinator_must_fail_an_arrival_from_an_unregistered_client_while_waiting()
+        {
+            var barrier = GetBarrier();
+            var a = CreateTestProbe();
+            var b = CreateTestProbe();
+            var stranger = CreateTestProbe();
+            barrier.Tell(new Controller.NodeInfo(A, Address.Parse("akka://sys"), a.Ref));
+            barrier.Tell(new Controller.NodeInfo(B, Address.Parse("akka://sys"), b.Ref));
+            a.Send(barrier, new EnterBarrier("bar12", null, A));
+
+            // A client the coordinator holds no registration for cannot be counted towards the
+            // barrier, so it has to be told that rather than left hanging on its ask.
+            stranger.Send(barrier, new EnterBarrier("bar12", null, C));
+            await stranger.ExpectMsgAsync(new ToClient<BarrierResult>(new BarrierResult("bar12", false)));
+
+            // The stray arrival leaves the barrier itself alone.
+            b.Send(barrier, new EnterBarrier("bar12", null, B));
+            await a.ExpectMsgAsync(new ToClient<BarrierResult>(new BarrierResult("bar12", true)));
+            await b.ExpectMsgAsync(new ToClient<BarrierResult>(new BarrierResult("bar12", true)));
+        }
+
         //TODO: Controller tests.
 
         private IActorRef GetBarrier()

src/core/Akka.Remote.TestKit.Tests/ControllerSpec.cs

@@ -7,9 +7,12 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Net;
+using System.Threading.Tasks;
 using Akka.Actor;
 using Akka.TestKit;
+using Akka.Util;
 using Xunit;
 
 namespace Akka.Remote.TestKit.Tests
@@ -48,6 +51,40 @@ public void Controller_must_publish_its_nodes()
                 ExpectMsg<Terminated>();
             }, TimeSpan.FromSeconds(20));
         }
+
+        [Fact(DisplayName = "Controller should keep a re-registered node when its previous connection reports a disconnect")]
+        public async Task Controller_must_keep_a_re_registered_node_when_the_previous_connection_disconnects()
+        {
+            var address = Address.Parse("akka://sys");
+            var c = Sys.ActorOf(Props.Create(() => new Controller(1, new IPEndPoint(IPAddress.Loopback, 0))));
+            var oldConnection = CreateTestProbe();
+            var newConnection = CreateTestProbe();
+
+            oldConnection.Send(c, new Controller.NodeInfo(A, address, oldConnection.Ref));
+            await oldConnection.ExpectMsgAsync<ToClient<Done>>();
+
+            // Tear the node down the way TestConductor.Shutdown does, then let it come back on a
+            // fresh connection under the same role, the way StartNewSystem does.
+            c.Tell(new Terminate(A, new Left<bool, int>(true)));
+            await oldConnection.ExpectMsgAsync<ToClient<TerminateMsg>>();
+
+            newConnection.Send(c, new Controller.NodeInfo(A, address, newConnection.Ref));
+            await newConnection.ExpectMsgAsync<ToClient<Done>>();
+
+            // The connection that has already been replaced now reports its disconnect. It must
+            // not evict the registration that replaced it.
+            oldConnection.Send(c, new Controller.ClientDisconnected(A));
+
+            c.Tell(Controller.GetNodes.Instance);
+            var nodes = await ExpectMsgAsync<IEnumerable<RoleName>>();
+            Assert.Contains(A, nodes.ToList());
+
+            // The barrier coordinator has to still know the node as well, otherwise its arrivals
+            // are ignored and every later barrier stalls.
+            newConnection.Send(c, new EnterBarrier("after-restart", null, A));
+            var result = await newConnection.ExpectMsgAsync<ToClient<BarrierResult>>();
+            Assert.True(result.Msg.Success);
+        }
     }
 }
 

src/core/Akka.Remote.TestKit/BarrierCoordinator.cs

@@ -556,9 +556,17 @@ private void InitFSM()
                     case EnterBarrier barrier:
                         if (barrier.Name != currentBarrier)
                             throw new WrongBarrierException(barrier.Name, Sender, barrier.Role, @event.StateData);
-                        var together = clients.Any(x => Equals(x.FSM, Sender))
-                            ? @event.StateData.Arrived.Add(Sender)
-                            : @event.StateData.Arrived;
+
+                        // An arrival from a client we hold no registration for can never be
+                        // counted towards this barrier, so failing it is the only honest answer.
+                        // Staying silent leaves that client blocked on an ask nothing will ever
+                        // complete: it reports a 60 second ask timeout long after the barrier
+                        // itself timed out on another node, which buries the real cause. Idle
+                        // already answers an unregistered sender this way.
+                        if (!clients.Any(x => Equals(x.FSM, Sender)))
+                            return Stay().Replying(new ToClient<BarrierResult>(new BarrierResult(barrier.Name, false)));
+
+                        var together = @event.StateData.Arrived.Add(Sender);
                         var enterDeadline = GetDeadline(barrier.Timeout);
                         //we only allow the deadlines to get shorter
                         if (enterDeadline.TimeLeft < @event.StateData.Deadline.TimeLeft)

src/core/Akka.Remote.TestKit/Conductor.cs

@@ -530,7 +530,10 @@ protected void InitFSM()
 
             OnTermination(_ =>
             {
-                _controller.Tell(new Controller.ClientDisconnected(_roleName));
+                // Name the sender explicitly. The controller compares it against the FSM it has
+                // registered for this role, which is how it tells a disconnect that belongs to
+                // this connection apart from one that a newer connection already replaced.
+                _controller.Tell(new Controller.ClientDisconnected(_roleName), Self);
                 _channel.CloseAsync();
             });
 

src/core/Akka.Remote.TestKit/Controller.cs

@@ -307,6 +307,19 @@ protected override void OnReceive(object message)
             var clientDisconnected = message as ClientDisconnected;
             if (clientDisconnected != null && clientDisconnected.Name != null)
             {
+                // A ServerFSM can report its disconnect after the same role has already come back
+                // on a fresh channel, which is exactly what a node that calls StartNewSystem does.
+                // Removing the role by name alone would drop the live registration, and from then
+                // on the barrier coordinator ignores every arrival from that node, so the next
+                // barrier stalls until it times out. Only let a ServerFSM evict the registration
+                // it owns.
+                if (_nodes.TryGetValue(clientDisconnected.Name, out var registered)
+                    && !Equals(registered.FSM, Sender))
+                {
+                    _log.Debug("Ignoring disconnect of superseded connection for {0}", clientDisconnected.Name);
+                    return;
+                }
+
                 _nodes = _nodes.Remove(clientDisconnected.Name);
                 _barrier.Forward(clientDisconnected);
                 return;

src/core/Akka.Remote.TestKit/Player.cs

@@ -177,7 +177,9 @@ public async Task EnterAsync(TimeSpan timeout, RoleName roleName, ImmutableList<
             {
                 var askTimeout = barrierTimeout + Settings.QueryTimeout;
                 // Use async ask with cancellation token
-                var result = await _client.Ask(new ToServer<EnterBarrier>(new EnterBarrier(name, barrierTimeout, roleName)), askTimeout, cancellationToken);
+                // A failed barrier now faults this ask (see the Status.Failure note in ClientFSM), so the
+                // exception propagates to the caller instead of the spec silently continuing unsynchronized.
+                await _client.Ask(new ToServer<EnterBarrier>(new EnterBarrier(name, barrierTimeout, roleName)), askTimeout, cancellationToken);
             }
             catch (TaskCanceledException ex)
             {
@@ -421,7 +423,7 @@ public void InitFSM()
                     _log.Error("Received {0} instead of Done", @event.FsmEvent);
                     return GoTo(State.Failed);
                 case IServerOp:
-                    return Stay().Replying(new Failure(new IllegalStateException("not connected yet")));
+                    return Stay().Replying(new Status.Failure(new IllegalStateException("not connected yet")));
                 case StateTimeout:
                     _log.Error("connect timeout to TestConductor");
                     return GoTo(State.Failed);
@@ -475,17 +477,22 @@ public void InitFSM()
                     else
                     {
                         object response;
+                        // NOTE: these MUST be Status.Failure. The unqualified name `Failure` binds to the
+                        // inherited FSMBase.Failure (an FSM termination reason), which the ask completion
+                        // switch in FutureActorRef does not treat as a fault - it falls through to
+                        // `case T t: TrySetResult(t)`, so the ask completes SUCCESSFULLY and the caller
+                        // walks through a barrier that never synchronized.
                         if (barrierResult.Name != @event.StateData.RunningOp.Value.Item1)
                         {
                             response =
-                                new Failure(
+                                new Status.Failure(
                                     new Exception("wrong barrier " + barrierResult + " received while waiting for " +
                                                   @event.StateData.RunningOp.Value.Item1));
                         }
                         else if (!barrierResult.Success)
                         {
                             response =
-                                new Failure(
+                                new Status.Failure(
                                     new Exception("barrier failed:" + @event.StateData.RunningOp.Value.Item1));
                         }
                         else

src/core/Akka.Remote.TestKit/RemoteConnection.cs

@@ -10,6 +10,7 @@
 using System.Linq;
 using System.Net;
 using System.Net.Sockets;
+using System.Threading;
 using System.Threading.Tasks;
 using Akka.Remote.TestKit.Proto;
 using Akka.Remote.Transport.DotNetty;
@@ -126,9 +127,17 @@ public static void Shutdown(IChannel connection)
 
         public static async Task ReleaseAll()
         {
-            Task tc = _clientPool?.ShutdownGracefullyAsync() ?? TaskEx.Completed;
-            Task ts = _serverPool?.ShutdownGracefullyAsync() ?? TaskEx.Completed;
-            await Task.WhenAll(tc, ts).ConfigureAwait(false);
+            // Detach the pools before shutting them down. A shut down event loop group cannot
+            // serve another connection, so leaving the fields set hands every later
+            // CreateConnection call a dead pool and the connection never completes.
+            var clientPool = Interlocked.Exchange(ref _clientPool, null);
+            var serverPool = Interlocked.Exchange(ref _serverPool, null);
+            var serverWorkerPool = Interlocked.Exchange(ref _serverWorkerPool, null);
+
+            Task tc = clientPool?.ShutdownGracefullyAsync() ?? TaskEx.Completed;
+            Task ts = serverPool?.ShutdownGracefullyAsync() ?? TaskEx.Completed;
+            Task tsw = serverWorkerPool?.ShutdownGracefullyAsync() ?? TaskEx.Completed;
+            await Task.WhenAll(tc, ts, tsw).ConfigureAwait(false);
         }
 
         #endregion