Masstransit: Support Continuous Retry on Initial Connection Failure

[phatboyg]

With the move to container-based deployment, and throwing RabbitMQ into a container with other services, it's possible that RabbitMQ will not be available at service start time.

Allow a timeout to be specified, during which the connection to RabbitMQ will be retried if it fails. This will allow time for the adjacent RabbitMQ container to start, and establish the connection.

[alexvy86]

Do you see any particular advantages of a timeout over just retrying indefinitely until the connection is established? That's what happens when the connection is already established and gets broken, right? My intention was to make the starting connection to behave the same.

[phatboyg]

Because infinite anything is always bad.

[alexvy86]

Fair enough. Do the retries when the connection gets broken also have a timeout? I just haven't seen that they do.

[jehof]

Great idea. We have a custom workaround for this case. But it would be nice to have it directly in Masstransit

[alexvy86]

@jehof what's the workaround you're using? I'm also looking for a good way to handle an initial connection failure while this feature makes it into MassTransit.

[jehof]

@alexvy86 We try to start the BusControl (StartAsync) using a do-while loop. If the connection fails we wait some seconds (5 seconds) and try it again. The loop will abort if establishing the connection fails 5-times.

[Jstall]

We ended up doing something similar to what @jehof did for our ASP.Net Core app. Since in application startup we are doing allot of dependency injection that needs IBusControl I created an extension method that uses a Polly resilience policy that I run before app startup:
Very roughly, something like:

public class Program
{
      public static void Main(string[] args)
      {
          BuildWebHost(args)
         .EnsureServiceBus()
         .Run();
       }
 }

[alexvy86]

So inside EnsureServiceBus() you have a loop that instantiates a bus, calls StartAsync, and tries both things again until it succeeds? And then add that bus instance to the DI container instead of letting it instantiate it on its own?

[Jstall]

@alexvy86 - yes that is essentially it. In the loop, before the application runs, we are doing the entire setup :

var bus = Bus.Factory.CreateUsingRabbitMq(..)
bus.Start();
bus.Stop();

process and catching RabbitMqConnectionException. This continues for a given number of tries over a set intraval and successful completion indicates the service is running and can accept connections

[alexvy86]

@phatboyg do you have any pointers regarding what happens (at a low level) when MassTransit starts a bus? I think I'll have some time this weekend to look into this, but couldn't make much headway last time, so anything you could tell me would help. I think I identified some of the components involved (ConnectionContextFactory, RabbitMqReceiveTransport) but I still don't have a clear idea of the order of things or how they interact with each other.

So far the only difference I recall seeing between the initial connection attempt and the retries when the connection is broken, is a property (either Complete or Ready, can't recall which one) of an Agent object somewhere, that made me think it wasn't "ready" during the initial connection, but it was at the point where an established connection breaks and attempts to reconnect start.

[mjalil]

@phatboyg I do not agree with you, infinite anything is NOT always bad. if it's bad why you infinitely try to reconnect after a disconnect !?. IMHO, let's delegate the developer to decide when infinite is good, and when it is bad.
I suggest a config option to let us infinitely retry to connect at the FIRST TIME or not

[kharrak]

+1 - Same issue here. It seems that creating a new instance of the bus and trying to Start it until it succeeds (indefinetly or with a timeout - depending on the case) is the only workaround for now.

[tabareh]

Same issue here. Check out the sample with docker here:
https://github.com/tabareh/netcore_masstransit_rabbitmq_container

[netbidev]

I have asked a question on stackoverflow about implementation of healthcheck. Check the answer, maybe it would be useful - Implement IHealthCheckPublisher in MassTransit for .Net Core 2.2

[akinix]

install package Polly first
``` c#
using System;
using System.IO;
using MassTransit.RabbitMqTransport;
using Microsoft.AspNetCore;
using Microsoft.AspNetCore.Hosting;
using Polly;
using Serilog;
using Serilog.Events;

namespace MyCompanyName.MyProjectName
{
public class Program
{
public static int Main(string[] args)
{
Log.Logger = new LoggerConfiguration()
.MinimumLevel.Debug()
.MinimumLevel.Override("Microsoft", LogEventLevel.Information)
.Enrich.FromLogContext()
.WriteTo.File(@"logs/log.txt", rollingInterval: RollingInterval.Day)
.WriteTo.Console()
.CreateLogger();

        try
        {
            Log.Information("Starting web host.");
            var retryRabbitMqPolicy =
                Policy
                    .Handle<RabbitMqConnectionException>()
                    .WaitAndRetry(6, retryAttempt => TimeSpan.FromSeconds(5), (exception, retryCount) =>
                    {
                        Log.Error(exception, "An exception occurred while connecting to the rabbitmq server.");
                    });

            retryRabbitMqPolicy.Execute(() =>
            {
                BuildWebHost(args).Run();
            });
            return 0;
        }
        catch (Exception ex)
        {
            Log.Fatal(ex, "Host terminated unexpectedly!");
            return 1;
        }
        finally
        {
            Log.CloseAndFlush();
        }
    }

    public static IWebHost BuildWebHostInternal(string[] args) =>
        new WebHostBuilder()
            .UseKestrel()
            .UseContentRoot(Directory.GetCurrentDirectory())
            .UseIISIntegration()
            .UseStartup<Startup>()
            .UseSerilog()
            .Build();
}

}
```

[phatboyg]

Adding a default 60-second timeout on StartAsync() if no CancellationToken is specified, if you specify your own cancellationToken it's on you to cancel it after a timeout of your own devices.

[rageshS]

I think there is no straightforward solution to solve this issue. I facing the same issue while i try to start the service. I am using MassTransit 5.5.4 and MassTransit.RabbitMQ 5.5.4

[phatboyg]

@rageshS if you pass a cancellationToken with a longer timeout, it will wait longer. THe default wait time is I think 30 or 60 seconds before cancelling.

[stevef51]

I have this problem too, at least for local development where I am using RabbitMQ, for production Im using AzureServiceBus which is always running before the app containers start - but when I run locally with Docker and all the services including RabbitMQ are started at the same time the .NET Core apps will sit in a loop retrying the RabbitMQ connection which is great - the problem is once RabbitMQ container is up & running the .NET Core apps will throw an AggregateException containing each of the failed connections (I presume), the app starts (ie HTTP starts listening) but the actual RabbitMQ connection is not showing in the RabbitMQ management console - if I start RabbitMQ beforehand and allow it to settle, then run my .NET Core apps it connects first time and the console shows the connection ..
Output from my .NET Core app when RabbitMQ container starts around [09:45:05] ..

[09:45:01 ERR] RabbitMQ Connect Failed: rabbitmq:5672/
MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: rabbitmq:5672/ ---> 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: Connection refused 172.18.0.5:5672
   at System.Net.Sockets.Socket.EndConnect(IAsyncResult asyncResult)
   at System.Net.Sockets.Socket.<>c.<ConnectAsync>b__272_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..ctor(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 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.RabbitMqTransport.Transport.RabbitMqReceiveTransport.<Receiver>b__12_0()
The thread 3513 has exited with code 0 (0x0).
[09:45:15 FTL] An error occurred starting the application
System.AggregateException: One or more errors occurred. (One or more errors occurred. (One or more errors occurred. (Broker unreachable: rabbitmq:5672/))) ---> System.AggregateException: One or more errors occurred. (One or more errors occurred. (Broker unreachable: rabbitmq:5672/)) ---> System.AggregateException: One or more errors occurred. (Broker unreachable: rabbitmq:5672/) ---> MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: rabbitmq:5672/ ---> 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: No such device or address
   at System.Net.Dns.InternalGetHostByName(String hostName)
   at System.Net.Dns.ResolveCallback(Object context)
--- End of stack trace from previous location where exception was thrown ---
   at System.Net.Dns.HostResolutionEndHelper(IAsyncResult asyncResult)
   at System.Net.Dns.EndGetHostAddresses(IAsyncResult asyncResult)
   at System.Net.Dns.<>c.<GetHostAddressesAsync>b__25_1(IAsyncResult asyncResult)
   at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- 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..ctor(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 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.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 MassTransit.AspNetCoreIntegration.MassTransitHostedService.StartAsync(CancellationToken cancellationToken)
   at Microsoft.AspNetCore.Hosting.Internal.HostedServiceExecutor.ExecuteAsync(Func`2 callback)
   --- End of inner exception stack trace ---
   at Microsoft.AspNetCore.Hosting.Internal.HostedServiceExecutor.ExecuteAsync(Func`2 callback)
   at Microsoft.AspNetCore.Hosting.Internal.HostedServiceExecutor.StartAsync(CancellationToken token)
---> (Inner Exception #0) System.AggregateException: One or more errors occurred. (One or more errors occurred. (Broker unreachable: rabbitmq:5672/)) ---> System.AggregateException: One or more errors occurred. (Broker unreachable: rabbitmq:5672/) ---> MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: rabbitmq:5672/ ---> 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: No such device or address
   at System.Net.Dns.InternalGetHostByName(String hostName)
   at System.Net.Dns.ResolveCallback(Object context)
--- End of stack trace from previous location where exception was thrown ---
   at System.Net.Dns.HostResolutionEndHelper(IAsyncResult asyncResult)
   at System.Net.Dns.EndGetHostAddresses(IAsyncResult asyncResult)
   at System.Net.Dns.<>c.<GetHostAddressesAsync>b__25_1(IAsyncResult asyncResult)
   at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- 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..ctor(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 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.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 MassTransit.AspNetCoreIntegration.MassTransitHostedService.StartAsync(CancellationToken cancellationToken)
   at Microsoft.AspNetCore.Hosting.Internal.HostedServiceExecutor.ExecuteAsync(Func`2 callback)
---> (Inner Exception #0) System.AggregateException: One or more errors occurred. (Broker unreachable: rabbitmq:5672/) ---> MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: rabbitmq:5672/ ---> 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: No such device or address
   at System.Net.Dns.InternalGetHostByName(String hostName)
   at System.Net.Dns.ResolveCallback(Object context)
--- End of stack trace from previous location where exception was thrown ---
   at System.Net.Dns.HostResolutionEndHelper(IAsyncResult asyncResult)
   at System.Net.Dns.EndGetHostAddresses(IAsyncResult asyncResult)
   at System.Net.Dns.<>c.<GetHostAddressesAsync>b__25_1(IAsyncResult asyncResult)
   at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- 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..ctor(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 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.Policies.PipeRetryExtensions.Retry(IRetryPolicy retryPolicy, Func`1 retryMethod, CancellationToken cancellationToken)
   --- End of inner exception stack trace ---
---> (Inner Exception #0) MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: rabbitmq:5672/ ---> 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: No such device or address
   at System.Net.Dns.InternalGetHostByName(String hostName)
   at System.Net.Dns.ResolveCallback(Object context)
--- End of stack trace from previous location where exception was thrown ---
   at System.Net.Dns.HostResolutionEndHelper(IAsyncResult asyncResult)
   at System.Net.Dns.EndGetHostAddresses(IAsyncResult asyncResult)
   at System.Net.Dns.<>c.<GetHostAddressesAsync>b__25_1(IAsyncResult asyncResult)
   at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- 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..ctor(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 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.Policies.PipeRetryExtensions.Retry(IRetryPolicy retryPolicy, Func`1 retryMethod, CancellationToken cancellationToken)<---
<---
<---
Hosting environment: Development
Content root path: /src
Now listening on: http://[::]:80
Application started. Press Ctrl+C to shut down.
The thread 3517 has exited with code 0 (0x0).

and the RabbitMQ console ..

yet if I start RabbitMQ first, wait for 20 seconds for it idle then start my .NET Core app ..

so I dont understand why it sits there retrying over and over until RabbitMQ is actually alive but then fails to connect anyway.

I am using the following Nuget packages :-

<PackageReference Include="MassTransit" Version="6.0.0" />
<PackageReference Include="MassTransit.AspNetCore" Version="6.0.0" />
<PackageReference Include="MassTransit.Azure.ServiceBus.Core" Version="6.0.0" />
<PackageReference Include="MassTransit.Extensions.Logging" Version="5.5.6" />
<PackageReference Include="MassTransit.RabbitMQ" Version="6.0.0" />

and the following code to setup MassTransit (using Rabbit or AzureServiceBus through config) ..

            services.Configure<MassTransitOptions>(configuration.GetSection("MassTransit"));

            services.AddMassTransit(serviceProvider =>
            {
                var massTransitOptions = serviceProvider.GetRequiredService<IOptions<MassTransitOptions>>().Value;

                Func<RabbitMQOptions, IBusControl> createRabbitMQBus = (opts) =>
                {
                    var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
                    {
                        cfg.Host(new Uri($"rabbitmq://{opts.Server}"), h =>
                        {
                            h.Username(opts.Username);
                            h.Password(opts.Password);
                        });

                        // Setup a "temp" endpoint for this service, the endpoint (exchange + queue) will auto delete when we disconnect
                        // and also messages sent to this endpoint are not persisted
                        cfg.ReceiveEndpoint($"temp-{serviceName}-{Guid.NewGuid()}", x =>
                        {
                            x.Durable = false;
                            x.AutoDelete = true;

                            configureTempEndpoint?.Invoke(x, serviceProvider);

                            // MultiTenant consumes this event to clear its internal cache of Tenants
                            x.Consumer<TenantConnectionsUpdatedConsumer>(serviceProvider);
                        });

                        configureBus?.Invoke(cfg, serviceProvider);
                    });

                    return bus;
                };

                Func<AzureServiceBusOptions, IBusControl> createAzureServiceBus = (opts) =>
                {
                    var bus = Bus.Factory.CreateUsingAzureServiceBus(cfg =>
                    {
                        var host = cfg.Host(new Uri($"sb://{opts.Namespace}.servicebus.windows.net"), h =>
                        {
                            h.SharedAccessSignature(s =>
                            {
                                s.KeyName = opts.KeyName;
                                s.SharedAccessKey = opts.SharedAccessKey;
                                s.TokenTimeToLive = opts.TokenTimeToLive;
                                s.TokenScope = Microsoft.Azure.ServiceBus.Primitives.TokenScope.Namespace;
                            });
                        });

                        // Setup a "temp" endpoint for this service, the endpoint (exchange + queue) will auto delete when we disconnect
                        // and also messages sent to this endpoint are not persisted
                        cfg.SubscriptionEndpoint<TenantConnectionsUpdated>($"{serviceName}-{DateTime.UtcNow.Ticks}-{new System.Random().Next(100000)}", x =>
                        {
                            x.AutoDeleteOnIdle = new TimeSpan(0, 5, 0);
                            // MultiTenant consumes this event to clear its internal cache of Tenants
                            x.Consumer<TenantConnectionsUpdatedConsumer>(serviceProvider);
                        });

                        configureBus?.Invoke(cfg, serviceProvider);
                    });

                    return bus;
                };

                switch (massTransitOptions.Transport)
                {
                    case MassTransitOptions.TransportType.RabbitMQ:
                        return createRabbitMQBus(massTransitOptions.RabbitMQ);

                    case MassTransitOptions.TransportType.AzureServiceBus:
                        return createAzureServiceBus(massTransitOptions.AzureServiceBus);

                    default:
                        throw new ApplicationException("MassTransit must be configured");
                }
            });

ideally, what I think I should do is detect the exception and terminate the .NET Core app, docker will restart it and it will most likely run fine (since RabbitMQ will likely be idle) - however I dont seem to be able to hook the exception (.NET Core newbie'ish)

Any help much appreciated

[phatboyg]

@stevef51 this was fixed in 6.0.1 - you should upgrade your packages :)

[lnaie]

Now it seems that it retries indefinitely. How do I change the policy? My API won't start because of the MT retrying.

[phatboyg]

If you're using the latest version, the bus start doesn't block, but you're expected to add the health check to know if your service is healthy.

This is documented: https://masstransit-project.com/usage/configuration.html#asp-net-core

[lnaie]

Interesting.
The health check is exactly the reason why I was asking.

Added it and it works after the upgrade of nuget. But now I have a bit of overlapping with the health checks provided by https://github.com/Xabaril/AspNetCore.Diagnostics.HealthChecks.

2 observations:

• I noticed that if the rabbitmq service is stopped after the API started, your health check continues to show me the state Healthy which is not correct.
• you are making use of HealthReport's Description and Data to tell about the errors, while the other library is using the Exception property.
• If the rabbitmq is not up when the API starts, then I get this (I have my simple json formatter):

{
  "status": "Unhealthy",
  "results": {
    "masstransit-bus": {
      "status": "Unhealthy",
      "error": "MassTransit bus is not ready: "
    },
    "masstransit-endpoint": {
      "status": "Healthy",
      "error": "All endpoints ready"
    },
    "mongodb": {
      "status": "Healthy",
      "error": null
    }
  }
}

Now... how do I disable your health check reports, if I would aim for that? :)

[phatboyg]

How do you disable them?

You don’t add them.

[bkaid]

@lnaie If you remove the line cfg.UseHealthCheck(provider); it should remove the mass transit healthcheck. But I wouldn't say they are duplicate checks of the RabbitMQ checks from the Xabaril library. The MassTransit health check is simply checking if the IBus is properly started which is an important component of your services health.

[lnaie]

If the line is commented out/removed then the health reports are present and the status is Unhealthy even when the rabbitmq is up and running. If the line is present, then it show Healthy even when rabbitmq is down.
And it looks like the health check provider is added by .AddMassTransitHostedService(); and that means there are always 2 health report entries: masstransit-bus, masstransit-endpoint.

Not sure what to think of this approach. I'm actually interested in the status of rabbitmq and not the sate of bus library.

[phatboyg]

well, if you're using the bus and the bus isn't healthy, that's a bad thing. I don't see how you could not _care_ about it.

[lnaie]

fair point. maybe I have expected more from the built-in health checks. I will make a nice stew of masstransit and other (e.g. rabbitmq) healtch checks. thanks :)