AutorecoveringConnection - Connection reestablished after a dispose() or an abort()

[alessnet]

Hello,

We have a software that stops its processing when the model or a consumer cancels. However after a while we started to see some orphan connections without any channel. After investigation, it seems that disposing the IConnection does not always release the underlying connection.

We often see this issue after a network disconnection or if we manually disconnects the client from the RabbitMQ Management website.

In AutorecoveringConnection.cs you can see that the variable "manuallyClosed" is not synchronized nor volatile. But this variable is used in the recovery thread (recoveryTaskFactory) which could see the old value (false).
Anyway, even if you synchronize this variable, you can still call "dispose" while the auto-recovery is already in progress. The excepted behaviour is to automatically close the connection just after the recovery ended.

Furthermore, the "Dispose" method can throw an exception "The AMQP operation was interrupted". In C# we generally do not except an exception in the Dispose (what should we do?). But in this case it seems that the reason (operation was interrupted) is not a valid reason to throw. If the connection cannot send the "abort" packet, it should not throw any exception in my opinion.

See the following testcase (the "IsOpen" property becomes "true" after the dispose).

Thank you

using System;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using RabbitMQ.Client;
using System.Threading.Tasks;
using System.Net.Sockets;
using System.Collections.Generic;
using System.Threading;
using RabbitMQ.Client.Exceptions;

namespace testbugrabbitmq
{
	[TestClass]
	public class UnitTest1
	{
		[TestMethod]
		public async Task TestMethod1()
		{
			ConnectionFactory factory = new ConnectionFactory();
			factory.AutomaticRecoveryEnabled = true;
			factory.RequestedConnectionTimeout = 5000;
			factory.TopologyRecoveryEnabled = true;
			factory.SocketReadTimeout = 10000;
			factory.SocketWriteTimeout = 10000;
			factory.ContinuationTimeout = TimeSpan.FromSeconds(10);

			// used to keep track of opened connection and simulate a network error
			factory.SocketFactory = addr => new TestTcpClientAdapter(this, new Socket(addr, SocketType.Stream, ProtocolType.Tcp) { NoDelay = true });
			factory.UserName = "testcase";
			factory.Password = "testcase";
			factory.Uri = "amqp://myserver/testcase";

			IConnection connection = factory.CreateConnection();
			Assert.IsTrue(connection.IsOpen);

			bool wasDisconnected = false;
			Exception disposeException = null;
			connection.ConnectionShutdown += (s, e) =>
			{
				Task.Run(() =>
				{
					// the dispose may throw because the connection is closed (which is a weird behaviour)
					connection.Dispose();
				})
				.ContinueWith(t =>
				{
					/* 
					 * even if the "dispose" throwed an exception, you can see with the
					 * debugger that manuallyClosed is set to true.
					 * So we do not except the auto-recovery to reconnect.
					 */
					Volatile.Write(ref wasDisconnected, connection.IsOpen == false);
					Volatile.Write(ref disposeException, t.Exception?.InnerException);
				});
			};

			// simulate a network error
			this.CloseAllConnections();

			// wait to see if the "IsOpen" changes
			await Task.Delay(TimeSpan.FromSeconds(11));

			Assert.IsTrue(Volatile.Read(ref wasDisconnected));
			Exception test = Volatile.Read(ref disposeException);
			Assert.IsTrue(test == null || (test is OperationInterruptedException && test.Message == "The AMQP operation was interrupted"));

			/* The connection was restored */
			Assert.IsTrue(connection.IsOpen);
		}

		private void CloseAllConnections()
		{
			List<TcpClientAdapter> clonedList;
			lock (this.connections)
			{
				clonedList = new List<TcpClientAdapter>(this.connections);
				this.connections.Clear();
			}

			foreach (var connection in clonedList)
				connection.Close();
		}

		private List<TcpClientAdapter> connections = new List<TcpClientAdapter>();
		private class TestTcpClientAdapter : TcpClientAdapter
		{
			private readonly UnitTest1 parent;
			public TestTcpClientAdapter(UnitTest1 p, Socket socket) : base(socket)
			{
				this.parent = p;
				lock (this.parent.connections)
				{
					this.parent.connections.Add(this);
				}
			}

			public override void Close()
			{
				try
				{
					base.Close();
				}
				finally
				{
					lock (this.parent.connections)
					{
						this.parent.connections.Remove(this);
					}
				}

			}
		}
	}
}

[michaelklishin]

What version of the library is used?

[michaelklishin]

The excepted behaviour is to automatically close the connection just after the recovery ended.

I don't think any of the clients "automatically" closes connections if it is called in the middle of recovery. This is a natural race condition between connection recovery and your own code. There is only so much this library can do about that — that's why automatic recovery can be opted out of (and had been opt in for nearly the entire time of its existence).

[michaelklishin]

I filed #291 because synchronizing internal state is certainly a good idea but the rest is a responsibility of your application. We won't be automagically closing connections right after recovery: this will be very confusing to some.

[michaelklishin]

We never recommended using Dispose to close connections (if you can point me to a piece of documentation that does it, I'd be happy to correct it). There is a separate method for that, Close, and its variation that ignores all exceptions, Abort. Close is a network round-trip to the server. Without it, the only way for RabbitMQ to know that a client connection was gone is via the heartbeat mechanism.

[alessnet]

Hi @michaelklishin

Thanks for your feedback.
I'm using the latest nugget (4.1.1).
I still have a couple of remarks/questions.

My testcase seems a little bit convoluted but keep in mind that it was design to reproduce a difficult race condition.
In a real word situation, you can have this connection leak with the following code:

using (IConnection connection = factory.CreateConnection())
{
	// do nothing special
	Thread.Sleep(20000);
}

It's very rare but it happend if your network fails.

If you happen to close the connection at the wrong time, the connection is kept open or reopened later. By the wrong time, I mean just before or right during a recovery (manuallyclose = true AND performingRecovery = true).
Since the recovery happens automatically in the background at any possible time, how do you expect the programmer to handle this easily?

We never recommended using Dispose to close connections (if you can point me to a piece of documentation that does it, I'd be happy to correct it)

Nobody bother to document the IDisposable interface because it has a well-known meaning in C#: the dispose releases the underlying resources and the object should not be used anymore after that (generally it does not throw an exception unlike Java). Why is this interface implemented if not used?
I'm certainly not the only one to use the "using" construct due to the IDisposable (in fact it's recommanded). I'm going to test the Close() methods as you suggested, but I don't think it resolves my original issue (exceptions were already ignored in the testcase).

Furthermore, if the connection seems opened when I call Dispose() It would have be nice for the library to try a "clean" abort first (yes there is a network round-trip). And then a close() if the connection is actually broken. Similar to the following code (I don't know if this code is safe):

try
{
	if (connection.IsOpen)
		connection.Abort();
}
catch (Exception)
{
}
finally
{
	connection.Close();
}

We won't be automagically closing connections right after recovery: this will be very confusing to some.

I do not agree. If you call dispose() you should not use the object anymore. So nobody would be confused because by definition nobody uses this connection anymore. In my opinion it's more confusing to see your connection reappear after a dispose(). Personnaly I was very confused when I saw opened connections without any channels in the managment interface.

So in a nutshell, I propose this:

• Synchronize access to manuallyClosed (ok we already agree on this)
• Optional: if a recovery is in progress, try to cancel it.
• After the recovery successfully ends, check for "manuallyClosed" (for instance in RunRecoveryEventHandlers). If it's true, then abort the connection with the original reason ("Connection close forced" for instance).

Regards,

Alessandro

[michaelklishin]

Please post questions to rabbitmq-users or Stack Overflow. RabbitMQ uses GitHub issues for specific actionable items engineers can work on, not questions. Thank you.

[alessnet]

Sorry I didn't mean to be such a bother. I saw an error, I read the source code, I made a test case and I proposed a solution, or at least hoped for a clarification. I honestly try to help,

I have no question.

Have a nice day

[michaelklishin]

@alessnet thank you for the report. We will see what can be better synchronized. I now recall we have #288 already fixed for 3.6.7 (and later series, of course), which is in a similar area.

Cheers.