Possible connection leak when trying to create first connection

[lukebakken]

Discussed in #1855

^{Originally posted by krisz2000 August 8, 2025}

Describe the bug

Hi,

I'm trying to implement a rabbitmq connection that uses the ConnectionFactory with AutomaticRecoveryEnabled set to true. I was using code similar to what the documentation says. To make automatic recovery work I want to make sure that the first connection is created properly, so when a connection attempt fails I retry again after a 5 second delay. To simulate a rabbitmq that is unreachable I use a docker container running the latest image (rabbitmq:4.1.3-management). I start my dotnet service that tries to create the first connection at startup and as expected it fails at first with this exception:

> RabbitMQ connection creation failure.
      RabbitMQ.Client.Exceptions.BrokerUnreachableException: None of the specified endpoints were reachable
       ---> System.AggregateException: One or more errors occurred. (Connection failed, host 127.0.0.1:5672)
       ---> RabbitMQ.Client.Exceptions.ConnectFailureException: Connection failed, host 127.0.0.1:5672
       ---> System.Net.Sockets.SocketException (111): Connection refused
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.ThrowException(SocketError error, CancellationToken cancellationToken)
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.System.Threading.Tasks.Sources.IValueTaskSource.GetResult(Int16 token)
         at System.Net.Sockets.Socket.<ConnectAsync>g__WaitForConnectWithCancellation|285_0(AwaitableSocketAsyncEventArgs saea, ValueTask connectTask, CancellationToken cancellationToken)
         at RabbitMQ.Client.Impl.SocketFactory.ConnectUsingAddressFamilyAsync(IPEndPoint endpoint, Func`2 socketFactory, AddressFamily family, TimeSpan connectionTimeout, CancellationToken cancellationToken)
         --- End of inner exception stack trace ---
         at RabbitMQ.Client.Impl.SocketFactory.ConnectUsingAddressFamilyAsync(IPEndPoint endpoint, Func`2 socketFactory, AddressFamily family, TimeSpan connectionTimeout, CancellationToken cancellationToken)
         at RabbitMQ.Client.Impl.SocketFactory.OpenAsync(AmqpTcpEndpoint amqpTcpEndpoint, Func`2 socketFactory, TimeSpan connectionTimeout, CancellationToken cancellationToken)
         at RabbitMQ.Client.Impl.SocketFrameHandler.CreateAsync(AmqpTcpEndpoint amqpTcpEndpoint, Func`2 socketFactory, TimeSpan connectionTimeout, CancellationToken cancellationToken)
         at RabbitMQ.Client.ConnectionFactory.CreateFrameHandlerAsync(AmqpTcpEndpoint endpoint, CancellationToken cancellationToken)
         at RabbitMQ.Client.EndpointResolverExtensions.SelectOneAsync[T](IEndpointResolver resolver, Func`3 selector, CancellationToken cancellationToken)
         --- End of inner exception stack trace ---
         at RabbitMQ.Client.EndpointResolverExtensions.SelectOneAsync[T](IEndpointResolver resolver, Func`3 selector, CancellationToken cancellationToken)
         at RabbitMQ.Client.Framing.AutorecoveringConnection.CreateAsync(ConnectionConfig config, IEndpointResolver endpoints, CancellationToken cancellationToken)
         at RabbitMQ.Client.ConnectionFactory.CreateConnectionAsync(IEndpointResolver endpointResolver, String clientProvidedName, CancellationToken cancellationToken)
         --- End of inner exception stack trace ---
         at RabbitMQ.Client.ConnectionFactory.CreateConnectionAsync(IEndpointResolver endpointResolver, String clientProvidedName, CancellationToken cancellationToken)

After I start the rabbitmq container again the following exception is thrown a couple of times

> RabbitMQConnection[0]
      RabbitMQ connection creation failure.
      RabbitMQ.Client.Exceptions.BrokerUnreachableException: None of the specified endpoints were reachable
       ---> System.IO.IOException: connection.start was never received, likely due to a network timeout
       ---> System.IO.EndOfStreamException: Pipe is completed.
         at RabbitMQ.Client.Impl.InboundFrame.ReadFromPipeAsync(PipeReader reader, UInt32 maxInboundMessageBodySize, InboundFrame frame, CancellationToken mainLoopCancellationToken)
         at RabbitMQ.Client.Framing.Connection.ReceiveLoopAsync(CancellationToken mainLoopCancellationToken)
         at RabbitMQ.Client.Framing.Connection.MainLoop()
         --- End of inner exception stack trace ---
         at RabbitMQ.Client.Framing.Connection.StartAndTuneAsync(CancellationToken cancellationToken)
         at RabbitMQ.Client.Framing.Connection.OpenAsync(CancellationToken cancellationToken)
         at RabbitMQ.Client.Framing.Connection.OpenAsync(CancellationToken cancellationToken)
         at RabbitMQ.Client.Framing.AutorecoveringConnection.CreateAsync(ConnectionConfig config, IEndpointResolver endpoints, CancellationToken cancellationToken)
         at RabbitMQ.Client.ConnectionFactory.CreateConnectionAsync(IEndpointResolver endpointResolver, String clientProvidedName, CancellationToken cancellationToken)
         --- End of inner exception stack trace ---
         at RabbitMQ.Client.ConnectionFactory.CreateConnectionAsync(IEndpointResolver endpointResolver, String clientProvidedName, CancellationToken cancellationToken)

The end result is that when I check the rabbitmq management frontend I see multiple connections have been created by my service. These connections are kept alive by automatic recovery, but I have no way to access them because the factory threw an exception while creating them.
Is this a bug or am I missing something?

Reproduction steps

1. Use the code found in the documentation for automatic connection recovery. In the retry logic try connecting again until a connection is created.
2. Start the service while the rabbitmq container is not running. Wait a bit for some connection failures.
3. Start the rabbitmq container
4. Wait until the connection is created
5. See that there are multiple connections in the rabbitmq management frontend.

Expected behavior

Only one rabbitmq connection should be created and kept alive by the factory

Additional context

Rabbitmq management version: 4.1.3
Rabbitmq dotnet client version: 7.1.2
Dotnet version: 8.0

https://github.com/lukebakken/rabbitmq-dotnet-client-1855

[krisz2000]

After a bit of debuging I managed to locate the source of the issue.

rabbitmq-dotnet-client/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.cs

Lines 92 to 102 in a267025

	internal static async ValueTask<AutorecoveringConnection> CreateAsync(ConnectionConfig config, IEndpointResolver endpoints,
	CancellationToken cancellationToken)
	{
	IFrameHandler fh = await endpoints.SelectOneAsync(config.FrameHandlerFactoryAsync, cancellationToken)
	.ConfigureAwait(false);
	Connection innerConnection = new(config, fh);
	AutorecoveringConnection connection = new(config, endpoints, innerConnection);
	await innerConnection.OpenAsync(cancellationToken)
	.ConfigureAwait(false);
	return connection;
	}

When in line 99 innerConnection.OpenAsync throws an exception the AutorecoveringConnection created on line 98 is not cleaned up, so it continues it's recovery loop. One solution that worked for me is to enclose the innerConnection.OpenAsync in a try-catch block and call connection.CloseAsync if any error happens, like so:

try
{
    await innerConnection.OpenAsync(cancellationToken)
        .ConfigureAwait(false);
}
catch
{
    try
    {
        await connection.CloseAsync(Constants.InternalError, "FailedOpen",
            InternalConstants.DefaultConnectionCloseTimeout, true,
            cancellationToken).ConfigureAwait(false);
    }
    catch { }
    throw;
}

I'm not sure that this is the best solution so I would like your input on this one.

[michaelklishin]

@krisz2000 the proposed solution seems reasonable to me. In particular if it plugs a possible connection leak :)

[krisz2000]

Thank you, I'll open a pull request then soon.

[lukebakken]

@krisz2000 I'm so glad you took a look at this issue!

I have a small change to suggest:

internal static async ValueTask<AutorecoveringConnection> CreateAsync(ConnectionConfig config, IEndpointResolver endpoints,
    CancellationToken cancellationToken)
{
    IFrameHandler fh = await endpoints.SelectOneAsync(config.FrameHandlerFactoryAsync, cancellationToken)
        .ConfigureAwait(false);
    Connection innerConnection = new(config, fh);
    AutorecoveringConnection connection = new(config, endpoints, innerConnection);
    try
    {
        await innerConnection.OpenAsync(cancellationToken)
            .ConfigureAwait(false);
    }
    catch
    {
        try 
        {
            await connection.CloseAsync(cancellationToken)
                .ConfigureAwait(false);
            await connection.DisposeAsync()
                .ConfigureAwait(false);
            }
        catch
        {
        }

        throw;
    }

    return connection;
}

I'll keep an eye out for your PR. Thank you SO MUCH for doing this work.

[krisz2000]

@lukebakken Thank you for the suggestion on also calling DisposeAsync.
For CloseAsync I was thinking about possibilities where the _innerConnection is still open after an exception is thrown, and that's why I used the same parameters for the CloseAsync call as it is done in Connection.OpenAsync, but I guess right now it's not really a possibility, so if you prefer calling it this way I'm fine with that.

[lukebakken]

Oh, indeed, your way is better!

[lukebakken]

If only 10% of open-source users were as willing as @krisz2000 to step up and fix something, free software would be so much better. Thank you @krisz2000!

[krisz2000]

Thank you for your kind words @lukebakken. I'm glad I could help!