Backport MNTR conductor reliability to v1.5 (#8431 barrier honesty + #8485 conductor fixes)

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