Merge pull request #1624 from rabbitmq/rabbitmq-dotnet-client-1623

Fix issue when recovery takes longer than recovery interval

Author: lukebakken

projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recovery.cs

@@ -240,15 +240,27 @@ await _innerConnection.AbortAsync(Constants.InternalError, "FailedAutoRecovery",
 
         private async ValueTask<bool> TryRecoverConnectionDelegateAsync(CancellationToken cancellationToken)
         {
+            Connection? maybeNewInnerConnection = null;
             try
             {
                 Connection defunctConnection = _innerConnection;
+
                 IFrameHandler fh = await _endpoints.SelectOneAsync(_config.FrameHandlerFactoryAsync, cancellationToken)
                     .ConfigureAwait(false);
-                _innerConnection = new Connection(_config, fh);
-                await _innerConnection.OpenAsync(cancellationToken)
+
+                maybeNewInnerConnection = new Connection(_config, fh);
+
+                await maybeNewInnerConnection.OpenAsync(cancellationToken)
                     .ConfigureAwait(false);
-                _innerConnection.TakeOver(defunctConnection);
+                maybeNewInnerConnection.TakeOver(defunctConnection);
+
+                /*
+                 * Note: do this last in case something above throws an exception during re-connection
+                 * We don't want to lose te old defunct connection in this case, since we have to take
+                 * over its data / event handlers / etc when the re-connect eventually succeeds.
+                 * https://github.com/rabbitmq/rabbitmq-dotnet-client/issues/1623 
+                 */
+                _innerConnection = maybeNewInnerConnection;
                 return true;
             }
             catch (Exception e)
@@ -260,6 +272,8 @@ await _innerConnection.OpenAsync(cancellationToken)
                     // Note: recordedEntities semaphore is _NOT_ held at this point
                     _connectionRecoveryErrorWrapper.Invoke(this, new ConnectionRecoveryErrorEventArgs(e));
                 }
+
+                maybeNewInnerConnection?.Dispose();
             }
 
             return false;

projects/Test/SequentialIntegration/TestConnectionRecovery.cs

@@ -266,10 +266,58 @@ public async Task TestShutdownEventHandlersRecoveryOnConnectionAfterDelayedServe
 
             await WaitAsync(shutdownLatch, WaitSpan, "connection shutdown");
             await WaitAsync(recoveryLatch, WaitSpan, "connection recovery");
+
             Assert.True(_conn.IsOpen);
             Assert.True(counter >= 1);
         }
 
+        [Fact]
+        public async Task TestShutdownEventHandlersRecoveryOnConnectionAfterTwoDelayedServerRestarts_GH1623()
+        {
+            const int restartCount = 2;
+            int counter = 0;
+            TimeSpan delaySpan = TimeSpan.FromSeconds(_connFactory.NetworkRecoveryInterval.TotalSeconds * 2);
+
+            AutorecoveringConnection aconn = (AutorecoveringConnection)_conn;
+
+            aconn.ConnectionRecoveryError += (c, args) =>
+            {
+                // Uncomment for debugging
+                // _output.WriteLine("[INFO] ConnectionRecoveryError: {0}", args.Exception);
+            };
+
+            aconn.ConnectionShutdown += (c, args) => Interlocked.Increment(ref counter);
+
+            Assert.True(_conn.IsOpen);
+
+            TaskCompletionSource<bool> recoveryLatch = null;
+
+            for (int i = 0; i < restartCount; i++)
+            {
+                if (i == (restartCount - 1))
+                {
+                    recoveryLatch = PrepareForRecovery(aconn);
+                }
+
+                try
+                {
+                    await StopRabbitMqAsync();
+                    await Task.Delay(delaySpan);
+                }
+                finally
+                {
+                    await StartRabbitMqAsync();
+                    // Ensure recovery has a chance to connect!
+                    await Task.Delay(delaySpan);
+                }
+            }
+
+            await WaitAsync(recoveryLatch, WaitSpan, "connection recovery");
+
+            Assert.True(aconn.IsOpen);
+            Assert.Equal(restartCount, counter);
+        }
+
         [Fact]
         public async Task TestUnblockedListenersRecovery()
         {