MassTransit dies after attempting to connect to unavailable RabbitMQ broker when host service is starting up

[mrclayman]

Is this a bug report?

Yes

Can you also reproduce the problem with the latest version?

Yes, using version 6.0.0 that's a few days old.

Environment

1. Operating system: Linux
2. Visual Studio version: None
3. Dotnet version: 3.1.100

Steps to Reproduce

1. Make sure RabbitMQ broker is stopped
2. Start the client service that connects to RabbitMQ and wait until it fails to connect and enters the retry loop
3. Start the RabbitMQ broker

Expected Behavior

The client service successfully connects to the RabbitMQ instance and enters the main loop

Actual Behavior

Once a connection attempt is made after availability of the RabbitMQ instance is restored, the client service terminates with AggregateException and this traceback:

info: Microsoft.AspNetCore.DataProtection.KeyManagement.XmlKeyManager[0]
      User profile is available. Using '/home/clay/.aspnet/DataProtection-Keys' as key repository; keys will not be encrypted at rest.
Unhandled exception. System.AggregateException: One or more errors occurred. (One or more errors occurred. (Broker unreachable: test_user@localhost:5672/send_test))
 ---> System.AggregateException: One or more errors occurred. (Broker unreachable: test_user@localhost:5672/send_test)
 ---> MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: test_user@localhost:5672/send_test
 ---> RabbitMQ.Client.Exceptions.BrokerUnreachableException: None of the specified endpoints were reachable
 ---> System.AggregateException: One or more errors occurred. (Connection failed)
 ---> RabbitMQ.Client.Exceptions.ConnectFailureException: Connection failed
 ---> System.Net.Internals.SocketExceptionFactory+ExtendedSocketException (111): Connection refused 127.0.0.1:5672
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw(Exception source)
   at System.Net.Sockets.Socket.EndConnect(IAsyncResult asyncResult)
   at System.Net.Sockets.Socket.<>c.<ConnectAsync>b__275_0(IAsyncResult iar)
--- End of stack trace from previous location where exception was thrown ---
   at RabbitMQ.Client.TcpClientAdapter.ConnectAsync(String host, Int32 port)
   at RabbitMQ.Client.Impl.TaskExtensions.TimeoutAfter(Task task, Int32 millisecondsTimeout)
   at RabbitMQ.Client.Impl.SocketFrameHandler.ConnectOrFail(ITcpClient socket, AmqpTcpEndpoint endpoint, Int32 timeout)
   --- End of inner exception stack trace ---
   at RabbitMQ.Client.Impl.SocketFrameHandler.ConnectUsingAddressFamily(AmqpTcpEndpoint endpoint, Func`2 socketFactory, Int32 timeout, AddressFamily family)
   at RabbitMQ.Client.Impl.SocketFrameHandler.ConnectUsingIPv4(AmqpTcpEndpoint endpoint, Func`2 socketFactory, Int32 timeout)
   at RabbitMQ.Client.Impl.SocketFrameHandler..ctor(AmqpTcpEndpoint endpoint, Func`2 socketFactory, Int32 connectionTimeout, Int32 readTimeout, Int32 writeTimeout)
   at RabbitMQ.Client.Framing.Impl.IProtocolExtensions.CreateFrameHandler(IProtocol protocol, AmqpTcpEndpoint endpoint, Func`2 socketFactory, Int32 connectionTimeout, Int32 readTimeout, Int32 writeTimeout)
   at RabbitMQ.Client.ConnectionFactory.CreateFrameHandler(AmqpTcpEndpoint endpoint)
   at RabbitMQ.Client.EndpointResolverExtensions.SelectOne[T](IEndpointResolver resolver, Func`2 selector)
   --- End of inner exception stack trace ---
   at RabbitMQ.Client.EndpointResolverExtensions.SelectOne[T](IEndpointResolver resolver, Func`2 selector)
   at RabbitMQ.Client.ConnectionFactory.CreateConnection(IEndpointResolver endpointResolver, String clientProvidedName)
   --- End of inner exception stack trace ---
   at RabbitMQ.Client.ConnectionFactory.CreateConnection(IEndpointResolver endpointResolver, String clientProvidedName)
   at RabbitMQ.Client.ConnectionFactory.CreateConnection(IList`1 hostnames, String clientProvidedName)
   at MassTransit.RabbitMqTransport.Integration.ConnectionContextFactory.CreateConnection(ISupervisor supervisor)
   --- End of inner exception stack trace ---
   at MassTransit.RabbitMqTransport.Integration.ConnectionContextFactory.CreateConnection(ISupervisor supervisor)
   at MassTransit.RabbitMqTransport.Integration.ConnectionContextFactory.CreateSharedConnection(Task`1 context, CancellationToken cancellationToken)
   at GreenPipes.Agents.PipeContextSupervisor`1.GreenPipes.IPipeContextSource<TContext>.Send(IPipe`1 pipe, CancellationToken cancellationToken)
   at GreenPipes.Agents.PipeContextSupervisor`1.GreenPipes.IPipeContextSource<TContext>.Send(IPipe`1 pipe, CancellationToken cancellationToken)
   at GreenPipes.Agents.PipeContextSupervisor`1.GreenPipes.IPipeContextSource<TContext>.Send(IPipe`1 pipe, CancellationToken cancellationToken)
   at MassTransit.Transports.TransportStartExtensions.OnTransportStartup[T](ReceiveEndpointContext context, ISupervisor`1 supervisor, CancellationToken cancellationToken)
   at MassTransit.RabbitMqTransport.Transport.RabbitMqReceiveTransport.<Receiver>b__12_0()
   at MassTransit.RabbitMqTransport.Transport.RabbitMqReceiveTransport.<Receiver>b__12_0()
   at MassTransit.RabbitMqTransport.Transport.RabbitMqReceiveTransport.<Receiver>b__12_0()
   at MassTransit.Policies.PipeRetryExtensions.Retry(IRetryPolicy retryPolicy, Func`1 retryMethod, CancellationToken cancellationToken)
   at MassTransit.Policies.PipeRetryExtensions.Retry(IRetryPolicy retryPolicy, Func`1 retryMethod, CancellationToken cancellationToken)
   --- End of inner exception stack trace ---
   --- End of inner exception stack trace ---
   at MassTransit.Transports.StartHostHandle.ReadyOrNot(IEnumerable`1 endpoints)
   at MassTransit.MassTransitBus.Handle.ReadyOrNot(Task`1 ready)
   at MassTransit.MassTransitBus.StartAsync(CancellationToken cancellationToken)
   at MassTransit.MassTransitBus.StartAsync(CancellationToken cancellationToken)
   at ConsumerService.EventBus.ConsumerBusControl.StartAsync(CancellationToken cancellationToken) in /home/clay/devel/tests/testmasstransit/ConsumerService/EventBus/ConsumerBusControl.cs:line 22
   at Microsoft.AspNetCore.Hosting.HostedServiceExecutor.ExecuteAsync(Func`2 callback, Boolean throwOnFirstFailure)
   at Microsoft.AspNetCore.Hosting.WebHost.StartAsync(CancellationToken cancellationToken)
   at Microsoft.AspNetCore.Hosting.WebHostExtensions.RunAsync(IWebHost host, CancellationToken token, String startupMessage)
   at Microsoft.AspNetCore.Hosting.WebHostExtensions.RunAsync(IWebHost host, CancellationToken token, String startupMessage)
   at Microsoft.AspNetCore.Hosting.WebHostExtensions.RunAsync(IWebHost host, CancellationToken token)
   at Microsoft.AspNetCore.Hosting.WebHostExtensions.Run(IWebHost host)
   at ConsumerService.Program.Main(String[] args) in /home/clay/devel/tests/testmasstransit/ConsumerService/Program.cs:line 17

Reproducible Demo

See the repro repository at https://github.com/mrclayman/testmasstransit.git

The contents are two separate services (producer/consumer). Using either of the two, I was able to reproduce the issue locally.

Thanks!

[phatboyg]

I was able to reproduce. Of course, I had to install .NET Core 3.1, create the new virtual host, the user, etc. But once I did everything needed to run your project, I was able to reproduce the issue.

[phatboyg]

Okay, so I've found the why. And I've put in a fix that works for RabbitMQ. The same problem exists in all of the brokers though, so I'm looking at how to solve it across the board. More to come.

[phatboyg]

I find it funny that the change to retry connections was put in to deal with the broker slowly starting up, and the service failing because the broker wasn't ready. And that the retry doesn't actually fix it. I wonder if it was fixed at one point, and then broken again. Anyway, the new approach should be cleaner anyway. And work for publishing-only buses as well.

[mrclayman]

Thank you very much for the quick analysis and response, @phatboyg . We use MassTransit with RabbitMQ in our dockerized microservice system and the RabbitMQ container is notorious for its slow startup so it's pretty much a given that the other services have to wait for it. The strange thing is that this issue did not occur with version 5.5.6 of MassTransit and .NET Core 3.0. Yesterday, I decided to migrate to the newest versions of .NET Core and MassTransit and ran into this. I am not sure what change on whose part caused this, though. I should have probably done the transition in stages to be able to better pinpoint the culprit.

[mrclayman]

I can confirm now that downgrade to MassTransit version 5.5.6 eliminates the issue. One thing I have noticed is that MassTransit 5.5.6 appears to handle the unavailability problem more gracefully by reporting errors like RabbitMQ Connect Failed: Broker unreachable: test_user@localhost:5672/send_test without actually printing out the entire call stack, which I would guess means the exception slips through the checks.

[phatboyg]

That helps, a lot actually. The fact it worked as expected in 5.5.6 tells me that I broke something, and it isn't something that goes back to 2018 as I suspected. I'll have to take a fresh look today at what changed between 5.5.6 and 6.0.

[phatboyg]

Of course, a lot changed. And yes, I've figured it out. I'm actually going to move retry to the ContextFactory(ies) as well, so that it is cleaner across all transports.

[tungphuong]

I am also facing the same issue on V6. Do we have any option to workaround before the V6.1 release?