Make JoinAsync and JoinSeedNodesAsync more robust by checking cluster…

… UP status (#6033) * Make JoinAsync and JoinSeedNodesAsync more robust by using an async state * Update how join state is being handled * Fix missing exception * Update unit test * Update API Verify list * update IsUp check * Change IsUp implementation to check SelfMember instead * Remove state handling code * Remove spec * Revert state changes Co-authored-by: Aaron Stannard <aaron@petabridge.com>
akkadotnet · Jul 27, 2022 · b1eb688 · b1eb688
1 parent 59cda86
commit b1eb688
Show file tree

Hide file tree

Showing 5 changed files with 38 additions and 11 deletions.
diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.Core.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.Core.verified.txt
@@ -29,6 +29,7 @@ namespace Akka.Cluster
         public Akka.Cluster.IDowningProvider DowningProvider { get; }
         public Akka.Remote.DefaultFailureDetectorRegistry<Akka.Actor.Address> FailureDetector { get; }
         public bool IsTerminated { get; }
+        public bool IsUp { get; }
         public Akka.Actor.Address SelfAddress { get; }
         public Akka.Cluster.Member SelfMember { get; }
         public System.Collections.Immutable.ImmutableHashSet<string> SelfRoles { get; }

diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.DotNet.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.DotNet.verified.txt
@@ -29,6 +29,7 @@ namespace Akka.Cluster
         public Akka.Cluster.IDowningProvider DowningProvider { get; }
         public Akka.Remote.DefaultFailureDetectorRegistry<Akka.Actor.Address> FailureDetector { get; }
         public bool IsTerminated { get; }
+        public bool IsUp { get; }
         public Akka.Actor.Address SelfAddress { get; }
         public Akka.Cluster.Member SelfMember { get; }
         public System.Collections.Immutable.ImmutableHashSet<string> SelfRoles { get; }

diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.Net.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.Net.verified.txt
@@ -29,6 +29,7 @@ namespace Akka.Cluster
         public Akka.Cluster.IDowningProvider DowningProvider { get; }
         public Akka.Remote.DefaultFailureDetectorRegistry<Akka.Actor.Address> FailureDetector { get; }
         public bool IsTerminated { get; }
+        public bool IsUp { get; }
         public Akka.Actor.Address SelfAddress { get; }
         public Akka.Cluster.Member SelfMember { get; }
         public System.Collections.Immutable.ImmutableHashSet<string> SelfRoles { get; }

diff --git a/src/core/Akka.Cluster.Tests/ClusterSpec.cs b/src/core/Akka.Cluster.Tests/ClusterSpec.cs
@@ -12,6 +12,7 @@
 using System.Threading.Tasks;
 using Akka.Actor;
 using Akka.Configuration;
+using Akka.Event;
 using Akka.TestKit;
 using Akka.TestKit.Extensions;
 using Akka.Util.Internal;

diff --git a/src/core/Akka.Cluster/Cluster.cs b/src/core/Akka.Cluster/Cluster.cs
@@ -264,11 +264,16 @@ public void Join(Address address)
         /// <returns>Task which completes, once current cluster node reaches <see cref="MemberStatus.Up"/> state.</returns>
         public Task JoinAsync(Address address, CancellationToken token = default)
         {
+            if (_isTerminated)
+                throw new ClusterJoinFailedException("Cluster has already been terminated");
+
+            if (IsUp)
+                return Task.CompletedTask;
+
             var completion = new TaskCompletionSource<NotUsed>(TaskCreationOptions.RunContinuationsAsynchronously);
 
-            var timeout = Settings.RetryUnsuccessfulJoinAfter ?? TimeSpan.FromSeconds(10);
             var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(token);
-            timeoutCts.CancelAfter(timeout);
+            timeoutCts.CancelAfter(Settings.SeedNodeTimeout);
             timeoutCts.Token.Register(() =>
             {
                 timeoutCts.Dispose();
@@ -281,10 +286,10 @@ public Task JoinAsync(Address address, CancellationToken token = default)
                 timeoutCts.Dispose();
                 completion.TrySetResult(NotUsed.Instance);
             });
-
+            
             Join(address);
 
-            return completion.Task.WithCancellation(token);
+            return completion.Task;
         }
 
         private Address FillLocal(Address address)
@@ -333,28 +338,35 @@ public void JoinSeedNodes(IEnumerable<Address> seedNodes)
         /// </summary>
         /// <param name="seedNodes">TBD</param>
         /// <param name="token">TBD</param>
-        public Task JoinSeedNodesAsync(IEnumerable<Address> seedNodes, CancellationToken token = default(CancellationToken))
+        public Task JoinSeedNodesAsync(IEnumerable<Address> seedNodes, CancellationToken token = default)
         {
+            if (_isTerminated)
+                throw new ClusterJoinFailedException("Cluster has already been terminated");
+
+            if (IsUp)
+                return Task.CompletedTask;
+
             var completion = new TaskCompletionSource<NotUsed>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var nodes = seedNodes.ToList();
 
             var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(token);
             timeoutCts.CancelAfter(Settings.SeedNodeTimeout);
             timeoutCts.Token.Register(() =>
             {
                 timeoutCts.Dispose();
                 completion.TrySetException(new ClusterJoinFailedException(
-                    $"Node has not managed to join the cluster using provided seed node addresses: {string.Join(", ", seedNodes)}."));
+                    $"Node has not managed to join the cluster using provided seed node addresses: {string.Join(", ", nodes)}."));
             });
 
             RegisterOnMemberUp(() =>
             {
                 timeoutCts.Dispose();
                 completion.TrySetResult(NotUsed.Instance);
             });
+
+            JoinSeedNodes(nodes);
 
-            JoinSeedNodes(seedNodes);
-
-            return completion.Task.WithCancellation(token);
+            return completion.Task;
         }
 
         /// <summary>
@@ -421,7 +433,10 @@ private Task LeaveSelf()
                 return leaveTask;
 
             // Subscribe to MemberRemoved events
-            _clusterDaemons.Tell(new InternalClusterAction.AddOnMemberRemovedListener(() => tcs.TrySetResult(null)));
+            _clusterDaemons.Tell(new InternalClusterAction.AddOnMemberRemovedListener(() =>
+            {
+                tcs.TrySetResult(null);
+            }));
 
             // Send leave message
             ClusterCore.Tell(new ClusterUserAction.Leave(SelfAddress));
@@ -451,7 +466,10 @@ public void Down(Address address)
         /// <param name="callback">The callback that is run whenever the current member achieves a status of <see cref="MemberStatus.Up"/></param>
         public void RegisterOnMemberUp(Action callback)
         {
-            _clusterDaemons.Tell(new InternalClusterAction.AddOnMemberUpListener(callback));
+            if (IsUp)
+                callback();
+            else
+                _clusterDaemons.Tell(new InternalClusterAction.AddOnMemberUpListener(callback));
         }
 
         /// <summary>
@@ -524,6 +542,11 @@ public ImmutableHashSet<string> SelfRoles
         /// </summary>
         public bool IsTerminated { get { return _isTerminated.Value; } }
 
+        /// <summary>
+        /// Determine whether the cluster is in the UP state.
+        /// </summary>
+        public bool IsUp => SelfMember.Status == MemberStatus.Up || SelfMember.Status == MemberStatus.WeaklyUp;
+
         /// <summary>
         /// The underlying <see cref="ActorSystem"/> supported by this plugin.
         /// </summary>