Masstransit: Consumer doesn't consume messages after RabbitMQ restart

[Vjake]

Is this a bug report?

Yes

Can you also reproduce the problem with the latest version?

Yes

Environment

1. Operating system: Windows 10 Pro
2. Visual Studio version: Community 2019
3. Dotnet version: .NET Core 3.1
   

   Steps to Reproduce

Steps:

1. Set ConcurrentMessageLimit=1 (for easily bug reproducing; I noticed that it is more easily reproducible by hands with 1 concurrent message limit).
2. Setup 3 RabbitMQ nodes (3.8 version) as a cluster (sync mode: ha-mode=all; ha-sync-mode: automatic), also they should be placed in docker for easy achieving of bug.
3. Setup any proxy service for accessing the RabbitMQ cluster (e.g. HAProxy).
4. Setup a consumer connection with RabbitMQ cluster via HAProxy (3).
5. Send over 60000 messages (it happens when RabbitMQ syncing for a long time).
6. Make a delay in a consume method (it can be equal 5 sec; it is important to get an error of already closed channel or "NACK failed").
7. Randomly kill and start RabbitMQ nodes while consuming (after some iterations of restarting RabbitMQ nodes and syncing consumer will stop to consume messages).

Expected Behavior

Consumer always reconnects to queue.

Actual Behavior

• Consumer doesn't reconnect to queue.
• At channel section in RabbitMQ management message: "... no channels ...".
• Last exceptions:

Message NACK failed: 6, Original Exception: System.Threading.Channels.ChannelClosedException: The channel has been closed.
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
System.Threading.Channels.ChannelClosedException: The channel has been closed.
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken)
   at MassTransit.RabbitMqTransport.Pipeline.RabbitMqReceiveLockContext.Faulted(Exception exception)
...
System.Threading.Channels.ChannelClosedException: The channel has been closed.
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.RabbitMqTransport.Pipeline.RabbitMqBasicConsumer.<>c__DisplayClass23_0.<<HandleBasicDeliver>b__0>d.MoveNext()

Reproducible Demo

(Paste the link to an example project and exact instructions to reproduce the issue.)

[Vjake]

Any thoughts? The bug is reproduced randomly and often (I don't remember that it was with netcore 2.x)

[phatboyg]

I haven't looked at it yet.

[phatboyg]

You need to post the consumer code, or at least describe it, because the call stack doesn't entirely make sense. Is your consumer also producing messages?

[phatboyg]

So, I tried to reproduce the failure to reconnect, but couldn't. I did consistently reproduce the channel closed exception.

Source Code

[08:42:02 ERR] Message NACK failed: 33, Original Exception: System.Threading.Channels.ChannelClosedException: The channel has been closed.
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
System.Threading.Channels.ChannelClosedException: The channel has been closed.
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken)
   at MassTransit.RabbitMqTransport.Pipeline.RabbitMqReceiveLockContext.Faulted(Exception exception)
[08:42:02 ERR] T-FAULT rabbitmq://localhost/event-listener dc3a0000-ebb8-e450-7f0c-08d88bcf7b44
System.Threading.Channels.ChannelClosedException: The channel has been closed.
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.RabbitMqTransport.Pipeline.RabbitMqBasicConsumer.<>c__DisplayClass23_0.<<HandleBasicDeliver>b__0>d.MoveNext()

Using MassTransit 7.0.6 directly from NuGet.

[Vjake]

@phatboyg I'm sorry, I’ve been busy last month; here is the code. I can reproduce the bug with it, may be I can give some additional logs?

[gokhangurbuz]

I have same problem. I use rabbit cluster. After Masstransit connected to any rabbitmq node. When I restart connecting node. Masstransit reconnecting successfully However, I tried restart second time. It doesn't reconnect. I used version 7.0.6 and 7.0.7. When I tried other versions. I noticed version 6.0.0 working properly. its always reconnect successfuly. @phatboyg

[phatboyg]

And 7.1.0-develop?

[gokhangurbuz]

I checked 7.1.0-develop.(3049,3066,3069,3072) at now. They have same problematic behaviour. @phatboyg

[phatboyg]

Unless I can reproduce it, I can't fix it. Can it be reproduced without the complex clustering setup? I wasn't able to reproduce it on my end.

[gokhangurbuz]

It seems to be reverseproxy related. I tried without haProxy. It works properly. But I have no idea how its working on v6.

[Vjake]

I've reproduced without haproxy as well

[Vjake]

@phatboyg I've written a couple scripts for reproducing of bug. Repository link.

Steps:

1. Execute init.sh [address of docker host] [10 if you are using ubuntu based on WSL] (example: init.sh localhost 10 - it will deploy whole cluster
2. Then run Example and when it will start consuming messages follow step 3
3. Execute make_bug.sh [address of docker host] [10 if you are using ubuntu based on WSL] (example: make_bug.sh localhost 10) - it will do mess and when it will stop working masstransit will be already got stuck on consuming

Consistently when I get an error there are next logs (see last logs expecially Reached the end of the stream):
logs.txt

[phatboyg]

That error in your log is the exact problem fixed that I mentioned, related to a faulted Nack due to the channel being closed. I'll see if I can reproduce this using your test repo on my end.

[Vjake]

I've got the same problem on 7.1.0-develop.3072: after log The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of the stream. Possible authentication failure. and The channel has been closed there is no consuming.
7.1.0-develop.3072.txt

I've pushed new branch with the latest develop version of masstransit in the test repo.

[phatboyg]

My latest run:

info: Consumer[0]
      Iteration `29`
info: Consumer[0]
      Iteration `29` closed
info: Consumer[0]
      Iteration `30`
info: MassTransit[0]
      Disconnect: admin@localhost:5672/
info: MassTransit[0]
      Disconnected: admin@localhost:5672/
info: MassTransit[0]
      Connect: admin@localhost:5672/
warn: MassTransit[0]
      Retrying 00:00:06.2240000: Broker unreachable: admin@localhost:5672/
MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: admin@localhost:5672/
 ---> 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
   at RabbitMQ.Client.Framing.Impl.Connection.StartAndTune()
   at RabbitMQ.Client.Framing.Impl.Connection.Open(Boolean insist)
   at RabbitMQ.Client.Framing.Impl.Connection..ctor(IConnectionFactory factory, Boolean insist, IFrameHandler frameHandler, String clientProvidedName)
   at RabbitMQ.Client.Framing.Impl.ProtocolBase.CreateConnection(IConnectionFactory factory, Boolean insist, IFrameHandler frameHandler, String clientProvidedName)
   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.<>c__DisplayClass6_0.<<CreateConnection>b__0>d.MoveNext()
   --- End of inner exception stack trace ---
   at MassTransit.RabbitMqTransport.Integration.ConnectionContextFactory.<>c__DisplayClass6_0.<<CreateConnection>b__0>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at MassTransit.Policies.PipeRetryExtensions.Retry[T](IRetryPolicy retryPolicy, Func`1 retryMethod, CancellationToken cancellationToken)
info: Consumer[0]
      Iteration `30` closed
fail: MassTransit.ReceiveTransport[0]
      T-FAULT rabbitmq://localhost/test dc3a0000-ebb8-e450-9576-08d8a0490673
System.Threading.Channels.ChannelClosedException: The channel has been closed.
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.Transports.ReceivePipeDispatcher.Dispatch(ReceiveContext context, ReceiveLockContext receiveLock)
   at MassTransit.RabbitMqTransport.Pipeline.RabbitMqBasicConsumer.<>c__DisplayClass23_0.<<HandleBasicDeliver>b__0>d.MoveNext()
info: MassTransit[0]
      Connect: admin@localhost:5672/
warn: MassTransit[0]
      Retrying 00:00:09.8220000: Broker unreachable: admin@localhost:5672/
MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: admin@localhost:5672/
 ---> 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
   at RabbitMQ.Client.Framing.Impl.Connection.StartAndTune()
   at RabbitMQ.Client.Framing.Impl.Connection.Open(Boolean insist)
   at RabbitMQ.Client.Framing.Impl.Connection..ctor(IConnectionFactory factory, Boolean insist, IFrameHandler frameHandler, String clientProvidedName)
   at RabbitMQ.Client.Framing.Impl.ProtocolBase.CreateConnection(IConnectionFactory factory, Boolean insist, IFrameHandler frameHandler, String clientProvidedName)
   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.<>c__DisplayClass6_0.<<CreateConnection>b__0>d.MoveNext()
   --- End of inner exception stack trace ---
   at MassTransit.RabbitMqTransport.Integration.ConnectionContextFactory.<>c__DisplayClass6_0.<<CreateConnection>b__0>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at MassTransit.Policies.PipeRetryExtensions.Retry[T](IRetryPolicy retryPolicy, Func`1 retryMethod, CancellationToken cancellationToken)
info: MassTransit[0]
      Connect: admin@localhost:5672/
warn: MassTransit[0]
      Retrying 00:00:16.8800000: Broker unreachable: admin@localhost:5672/
MassTransit.RabbitMqTransport.RabbitMqConnectionException: Broker unreachable: admin@localhost:5672/
 ---> 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
   at RabbitMQ.Client.Framing.Impl.Connection.StartAndTune()
   at RabbitMQ.Client.Framing.Impl.Connection.Open(Boolean insist)
   at RabbitMQ.Client.Framing.Impl.Connection..ctor(IConnectionFactory factory, Boolean insist, IFrameHandler frameHandler, String clientProvidedName)
   at RabbitMQ.Client.Framing.Impl.ProtocolBase.CreateConnection(IConnectionFactory factory, Boolean insist, IFrameHandler frameHandler, String clientProvidedName)
   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.<>c__DisplayClass6_0.<<CreateConnection>b__0>d.MoveNext()
   --- End of inner exception stack trace ---
   at MassTransit.RabbitMqTransport.Integration.ConnectionContextFactory.<>c__DisplayClass6_0.<<CreateConnection>b__0>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at MassTransit.Policies.PipeRetryExtensions.Retry[T](IRetryPolicy retryPolicy, Func`1 retryMethod, CancellationToken cancellationToken)
info: MassTransit[0]
      Connect: admin@localhost:5672/
info: Consumer[0]
      Iteration `31`
info: Consumer[0]
      Iteration `31` closed
info: Consumer[0]
      Iteration `32`

[phatboyg]

All I did was move the BusRunner ahead of the Sender. You were publishing before the bus was started. I also added a bus health check to the sender so that the bus was healthy before starting to publish. I can show you the changes to your example, or create a PR to it.

        public async Task StartAsync(CancellationToken cancellationToken)
        {
            await WaitForHealthyBus(cancellationToken);

            _logger.LogInformation("Pushing...");
            for (var i = 1; i < 1000; i++)
                await _bus.Publish(new TestMessage
                {
                    A1 = Guid.NewGuid(),
                    A2 = Guid.NewGuid(),
                    A3 = Guid.NewGuid(),
                    A4 = Guid.NewGuid(),
                });
            ;
            _logger.LogInformation("Pushing has ended");
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }

        async Task WaitForHealthyBus(CancellationToken cancellationToken)
        {
            HealthResult result;
            do
            {
                result = _busHealth.CheckHealth();

                await Task.Delay(100, cancellationToken);
            } while (result.Status != BusHealthStatus.Healthy);
        }

[Vjake]

Hmm, I set sender ahead of bus to fullfill queue, after this sender is disposed and I get bug on bus only. If you look at my last log there are similiar error logs but after some iterations of connection/reconnection/error messages there will be last messages like The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of the stream. Possible authentication failure and that's all.

Can you push your fixed code at different branch? I'll try to reproduce bug with fixed code

[phatboyg]

Like I said, you shouldn't publish until the bus is started. I'll create a PR with my changes.

Also, your RabbitMQ cluster is unhealthy on my test, the nodes never show up on the management console together.

[phatboyg]

PR created.

[Vjake]

I'll check.

Interesting, I have all nodes in the management UI together after executing init.sh.

[phatboyg]

oh, I didn't run init.sh

I did reproduce the issue prior to my changes, FYI.

[Vjake]

I reproduced the bug with your PR (waited when sender ends its work and started make_bug.sh). The bug wasn't so easy to reproduce...
logs_848.txt

Again last msg AMQP close-reason, initiated by Library and The channel has been closed

By the way what will problems be if I use sending (or publishing) without starting of bus?

update For the second time it required 12 iterations

[phatboyg]

Without starting the bus, it doesn't have an actual sustained bus connection. Calls to send/publish don't fail prior to the start because of historical reasons - at some point I need to just block any calls to send/publish until the bus has been started.

[Vjake]

Thank you for explanation.
Due to these problems: I have a microservice that just needs to publish messages after processing some requests from another source. What is a correct way to setup the bus? (with endpoint it skips msgs in the queue, without it doesn't start; maybe I am missing something).

[phatboyg]

Always start the bus. Once started, it can be used.

[Vjake]

Ah, okay, got it.

---

I noticed in the logs that when masstransit raises this exception:

      Connection Failed: rabbitmq://localhost/test
MassTransit.RabbitMqTransport.RabbitMqConnectionException: ReceiveTransport faulted: admin@localhost:5672/
 ---> RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of the stream. Possible authentication failure.
   at RabbitMQ.Client.Impl.InboundFrame.ReadFrom(Stream reader, Byte[] frameHeaderBuffer)
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoopIteration()
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoop()
   at RabbitMQ.Client.Framing.Impl.Connection.EnsureIsOpen()
   at RabbitMQ.Client.Framing.Impl.Connection.CreateModel()
   at MassTransit.Util.ChannelExecutor.SynchronousFuture`1.Run()
--- End of stack trace from previous location where exception was thrown ---
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken)
   at MassTransit.RabbitMqTransport.Contexts.RabbitMqConnectionContext.CreateModel(CancellationToken cancellationToken)
   at MassTransit.RabbitMqTransport.Contexts.RabbitMqConnectionContext.MassTransit.RabbitMqTransport.ConnectionContext.CreateModelContext(CancellationToken cancellationToken)
   at MassTransit.Internals.Extensions.SupervisorExtensions.CreateAgentPipe`2.Send(T context)
   at GreenPipes.Internals.Extensions.TaskExtensions.<>c__DisplayClass5_0`1.<<OrCanceled>g__WaitAsync|0>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at MassTransit.RabbitMqTransport.Integration.ModelContextFactory.CreateSharedModel(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.ReceiveTransport`1.ReceiveTransportAgent.<Run>b__8_0()
   --- End of inner exception stack trace ---

it always gets stuck on consuming messages and doesn't reconnect again.

This exception appears once at the end of the logs and after it there are one or few messages of consuming, a log of an exception The channel has been closed and nothing else happens.

[Vjake]

I've tested a new version: the same problem 7.1.0-develop.3078
3078.txt

[thomastomanek]

We've found this when UAT'ing an upgrade to MT7.

I've managed to reproduce without a cluster/proxy (I was confused as to why that was necessary).

Code to repro is https://gist.github.com/thomastomanek/1797c689d6f76f961a9835fb327b2541
Just stop the broker when the program has started processing messages. It fails 100% of the time for me (taking the sleep out I haven't managed to reproduce).

I've repro'd this against the latest develop branch. Confirmed as per above comments that you have to have a message being processed when the connection dies, so that the ACK fails.

It seems as though the retry policy in ReceiveTransportAgent never kicks in because the call to _supervisor.Send doesn't complete.

async Task Run()
{
    while (!IsStopping)
    {
        await _retryPolicy.Retry(async () =>
        {
            try
            {
                _supervisor = _supervisorFactory();

                await _context.OnTransportStartup(_supervisor, Stopping).ConfigureAwait(false);

                if (IsStopping)
                    return;

                await _supervisor.Send(_transportPipe, Stopped).ConfigureAwait(false);
            }
            catch (ConnectionException exception)
            {
                await NotifyFaulted(exception).ConfigureAwait(false);
                throw;
            }
            catch (OperationCanceledException exception)
            {
                throw await NotifyFaulted(exception, "ReceiveTransport canceled: ").ConfigureAwait(false);
            }
            catch (Exception exception)
            {
                throw await NotifyFaulted(exception, "ReceiveTransport faulted: ").ConfigureAwait(false);
            }
        }, Stopping).ConfigureAwait(false);
    }
}

[phatboyg]

Again, same problem, you aren't waiting for the bus to be ready before publishing.

[thomastomanek]

I'm not sure what you mean in this case. If I split up the code into two separate processes, one that publishes and one that just has a receive endpoint for TestMessage, on a broker restart the publisher reconnects and resumes publishing fine. The consumer on the other hand never reconnects. I've also just noticed as well that the exchange binding has disappeared routing the message to the receive endpoint (although it does have zero consumers as well so a bit moot).

[phatboyg]

I'll look at it.

[thomastomanek]

Fixed. Thanks for the fast turnaround.

[phatboyg]

Did you already try it with the latest develop packages? I don't even see them on NuGet yet.

[phatboyg]

@Vjake with the latest develop packages, do you still see the issue?

[Vjake]

Hmm, I tried 7.1.3-develop.3194 branch and after log

RabbitMQ.Client.Exceptions.OperationInterruptedException: The AMQP operation was interrupted: AMQP close-reason, initiated
 by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of
 the stream. Possible authentication failure.

it seems that masstransit doesn't try to reconnect properly

3194_bug_log.txt

[phatboyg]

So, looking at that log, it seems like it actually reconnected and continued processing. The last few lines are continued consumption of messages.

[Vjake]

Yes, but actually rabbitmq doesn't see any connected consumer and consuming is stopped got stuck

[phatboyg]

@Vjake I've had make_bug.sh running for a long time now. Am I right that it just sits there and waits for channels to exist on the broker, and then restarts it again. Will it eventually stop? So far, Example has reconnected every time.

[Vjake]

Yes, it checks that there are no channels left for 100 seconds and when nobody reconnects it will stop; otherwise it starts a new iteration of hard rebooting haproxy and rabbitmqs (meanwhile making sure that the cluster is stable to ensure no data is lost).

Have you got a The AMQP operation was interrupted: AMQP close-reason, initiate by Library, code=0, text='End of stream' log? (I noticed that every time when consumer doesn't reconnect eventually this message appears once. Last time it had approximately taken an hour before message appeared and consumer stopped to reconnect)

I'm thinking about reproducing of the bug with the develop branch of masstransit repo; maybe it will give more information...

[phatboyg]

I changed it to project references and built it directly from source with the project linked, and it's been running for 2+ hours without any issues. Reconnects each time after the five (why five?) recycles of RabbitMQ each time it sees connected sessions.

[thomastomanek]

yeah I tried it, I synced my fork from develop. All looks fine to me

[Vjake]

Hmm, I also now built from the source with the project linked (latest develop branch) and after 63 iteration of consuming in Example I got the error log that I mentioned above and consumer stopped to consume...

what is the difference...

ps five was a random number

[phatboyg]

No clue as to what could be the difference. Mine is over 180 iterations.

[phatboyg]

Also, I haven't seen the stream error at all, at least not when I search the console window history. I haven't been watching the entire time.

[Vjake]

It seems I got it - how to reproduce the bug without make_bug.sh.
Somehow after first connection error (while masstransit tries to declare rabbitmq queue) and subsequent first successful reconnection - second successful reconnection won't happen (it is already different issue or maybe this one)

Steps:

1. Start consuming
2. Set breakpoint here
3. stop haproxy (or rabbitmqs)
4. wait until masstransit loses connection
5. start haproxy (or rabbitmqs)
6. when breakpoint activates stop haproxy and after haproxy is stopped - continue execution
7. after the error occured start haproxy
8. first time it reconnects correctly
9. again stop haproxy
10. wait until masstransit loses connection
11. start haproxy

[Vjake]

I think it is hardware difference maybe - timings

[phatboyg]

Hmm, so it's within the ConfigureTopologyFilter where the queue is declared – I wonder if that is somehow related to the issue since that's a logic gate. Hmm.

[Vjake]

Yeah, it is different. I can open a new one...

[phatboyg]

Open a new what? Issue, it's the root cause I think. I just got it with the debugger, and .NET threw an unhandled exception error, so this is somehow related to an uncaught exception.

/Applications/Rider.app/Contents/lib/ReSharperHost/macos-x64/dotnet/dotnet exec /Applications/Rider.app/Contents/lib/ReSharperHost/JetBrains.Debugger.Worker.exe --mode=client --frontend-port=59304 --plugins=
info: MassTransit[0]
      Connect: admin@localhost:5672/
info: Sender[0]
      Pushing...
info: Consumer[0]
      Iteration `1`
info: MassTransit[0]
      Disconnect: admin@localhost:5672/
info: MassTransit[0]
      Disconnected: admin@localhost:5672/
fail: MassTransit[0]
      S-FAULT rabbitmq://localhost/test dc3a0000-ebb8-e450-222c-08d8b43bdc13 Example.TestMessage
info: MassTransit[0]
      Connect: admin@localhost:5672/
info: Consumer[0]
      Iteration `1` closed
Unhandled exception. MassTransit.RabbitMqTransport.MessageNotConfirmedException: exchange:test => The message was not confirmed: End of stream
   at MassTransit.RabbitMqTransport.Integration.BatchPublisher.<>c__DisplayClass8_0.<<Publish>g__PublishAsync|0>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
   at MassTransit.RabbitMqTransport.Transport.RabbitMqSendTransport.SendPipe`1.Send(ModelContext modelContext)
   at MassTransit.RabbitMqTransport.Transport.RabbitMqSendTransport.SendPipe`1.Send(ModelContext modelContext)
   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 Example.Sender.StartAsync(CancellationToken cancellationToken)
   at Microsoft.Extensions.Hosting.Internal.Host.StartAsync(CancellationToken cancellationToken)
   at Microsoft.Extensions.Hosting.HostingAbstractionsHostExtensions.RunAsync(IHost host, CancellationToken token)
   at Microsoft.Extensions.Hosting.HostingAbstractionsHostExtensions.RunAsync(IHost host, CancellationToken token)
   at Example.Program.Main(String[] args)
   at Example.Program.<Main>(String[] args)

[Vjake]

a new issue

---

wait until 'Example.Sender' will end (or disable 'Example.Sender') its work and only then follow steps above
(sorry, forgot about sender - I have disabled it)

[phatboyg]

Right, I added a try/catch around Publish to fix that.

Also, this one is it I think:

fail: MassTransit[0]
      Connection Failed: rabbitmq://localhost/test
MassTransit.RabbitMqTransport.RabbitMqConnectionException: ReceiveTransport faulted: admin@localhost:5672/
 ---> RabbitMQ.Client.Exceptions.OperationInterruptedException: The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of the stream. Possible authentication failure.
   at RabbitMQ.Client.Impl.InboundFrame.ReadFrom(Stream reader, Byte[] frameHeaderBuffer)
   at RabbitMQ.Client.Impl.SocketFrameHandler.ReadFrame()
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoopIteration()
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoop()
   at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)
   at RabbitMQ.Client.Impl.ModelBase.ModelRpc(MethodBase method, ContentHeaderBase header, Byte[] body)
   at RabbitMQ.Client.Framing.Impl.Model._Private_ExchangeDeclare(String exchange, String type, Boolean passive, Boolean durable, Boolean autoDelete, Boolean internal, Boolean nowait, IDictionary`2 arguments)
   at RabbitMQ.Client.Impl.ModelBase.ExchangeDeclare(String exchange, String type, Boolean durable, Boolean autoDelete, IDictionary`2 arguments)
   at MassTransit.RabbitMqTransport.Contexts.RabbitMqModelContext.<>c__DisplayClass16_0.<MassTransit.RabbitMqTransport.ModelContext.ExchangeDeclare>b__0() in /Users/chris/Code/MassTransit/src/Transports/MassTransit.RabbitMqTransport/Contexts/RabbitMqModelContext.cs:line 100
   at MassTransit.Util.ChannelExecutor.<>c__DisplayClass10_0.<Run>g__RunMethod|0() in /Users/chris/Code/MassTransit/src/MassTransit/Util/ChannelExecutor.cs:line 123
   at MassTransit.Util.ChannelExecutor.<>c__DisplayClass10_0.<Run>b__1() in /Users/chris/Code/MassTransit/src/MassTransit/Util/ChannelExecutor.cs:line 128
   at MassTransit.Util.ChannelExecutor.SynchronousFuture`1.Run() in /Users/chris/Code/MassTransit/src/MassTransit/Util/ChannelExecutor.cs:line 238
--- End of stack trace from previous location where exception was thrown ---
   at MassTransit.Util.ChannelExecutor.Run[T](Func`1 method, CancellationToken cancellationToken) in /Users/chris/Code/MassTransit/src/MassTransit/Util/ChannelExecutor.cs:line 137
   at MassTransit.RabbitMqTransport.Pipeline.ConfigureTopologyFilter`1.ConfigureTopology(ModelContext context) in /Users/chris/Code/MassTransit/src/Transports/MassTransit.RabbitMqTransport/Pipeline/ConfigureTopologyFilter.cs:line 50
   at MassTransit.RabbitMqTransport.Pipeline.ConfigureTopologyFilter`1.<>c__DisplayClass3_0.<<GreenPipes-IFilter<MassTransit-RabbitMqTransport-ModelContext>-Send>b__0>d.MoveNext() in /Users/chris/Code/MassTransit/src/Transports/MassTransit.RabbitMqTransport/Pipeline/ConfigureTopologyFilter.cs:line 33
--- End of stack trace from previous location where exception was thrown ---
   at GreenPipes.PipeExtensions.OneTimeSetup[T](PipeContext context, Func`2 setupMethod, PayloadFactory`1 payloadFactory)
   at MassTransit.RabbitMqTransport.Pipeline.ConfigureTopologyFilter`1.GreenPipes.IFilter<MassTransit.RabbitMqTransport.ModelContext>.Send(ModelContext context, IPipe`1 next) in /Users/chris/Code/MassTransit/src/Transports/MassTransit.RabbitMqTransport/Pipeline/ConfigureTopologyFilter.cs:line 31
   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.ReceiveTransport`1.ReceiveTransportAgent.<Run>b__8_0() in /Users/chris/Code/MassTransit/src/MassTransit/Transports/ReceiveTransport.cs:line 127
   --- End of inner exception stack trace ---

[Vjake]

Yeah, this one is root problem as well as any exception in any called method at ConfigureTopology method I think (as I mentioned above I got an exception (that leads to problems with reconnections) from called Declare(ModelContext context, Queue queue) method)

[phatboyg]

I think I got it. I think what's happening is the OneTimeSetup closes with the exception, and then retries it forever once it's reconnected and gets the immediate fault. The fault should not initialize the OneTimeSetup. No, that isn't it why is the message being published to test instead of the message type?? F, that's annoying trying to catch exchange declares when they're all the same.

[phatboyg]

Okay, now I think I have it. I found the issue, at least I think so. Committing to GitHub but it's a simple change.

[phatboyg]

@Vjake pushed to GitHub, pull down the change and see if you can reproduce it now.

[phatboyg]

Well clearly that wasn't the right way to fix it.

[phatboyg]

So this is the line that is the reason for the issue. It causes the retry loop to throw, which for some reason needs to exist. And it only comes into play after the connection is established and then lost (which is why the breakpoint on declare exchange causes it to happen), because it has to go into the Attempt method for this throw to be reached.

HOWEVER. If I simply add a return after the await Attempt(), it causes a bunch of the unit tests to fail for some reason.

I'm tired, I give up for tonight, I have no idea why this happens. I reverted all my attempts to fix it because they broke the unit tests so badly.

[phatboyg]

So, I'm calling this done as I could consistently reproduce it and now with the final latest develop build it is no longer an issue.

Total time spent resolving it? Easily 12-16 hours in the past 1.5 days.

[Vjake]

Okay, thank you!

[phatboyg]

Seriously, thank you for the test rig to get it reproduced. That was a real tricky one. And the root cause turned out to be a one line change, though I cleaned it all up to make it easier to understand.