Masstransit: System.InvalidOperationException: The method 'OnMessage' or 'OnMessageAsync' has already been called.

[sharpjs]

Is this a bug report?

Yes. Looks like a recurrence of #747 's first issue.

Can you also reproduce the problem with the latest version?

We experienced the problem in 5.1.0. We have deployed 5.1.2 and are waiting to see if the problem occurs again.

Environment

• .NET 4.7.1
• Console EXEs and ASP.NET MVC web app on IIS
• Built with Visual Studio 2017 15.7.2
• Running on Azure App Service x64
• MassTransit using Azure Service Bus
• 3 processes connected to bus
• Each process receives some broadcast messages via per-process queues
• One process receives competing-consumer messages via a shared queue

Steps to Reproduce

Wait until problem occurs. No specific trigger is known.

Expected Behavior

MassTransit remains connected to Azure Service Bus indefinitely.

Actual Behavior

An exception from a Task is unobserved and rethrown from the finalizer thread. Afterward, MassTransit disconnects from Azure Service Bus and receives no more messages until the process is restarted. This occurs around the same time (within 1hr 30min) for all 3 of our processes connected to the bus.

Reproducible Demo

Would love to have one myself.

Stack Trace

System.AggregateException: A Task's exception(s) were not observed either by Waiting on the Task or accessing its Exception property. As a result, the unobserved exception was rethrown by the finalizer thread.
System.InvalidOperationException: The method 'OnMessage' or 'OnMessageAsync' has already been called.
   at Microsoft.ServiceBus.Messaging.MessageReceiver.OnMessage (Microsoft.ServiceBus, Version=3.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35)
   at Microsoft.ServiceBus.Messaging.QueueClient.OnMessageAsync (Microsoft.ServiceBus, Version=3.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35)
   at MassTransit.AzureServiceBusTransport.Transport.Receiver.Start (MassTransit.AzureServiceBusTransport, Version=5.1.0.1516, Culture=neutral, PublicKeyToken=b8e0e9f2f1e657fa)
   at MassTransit.AzureServiceBusTransport.Pipeline.MessageReceiverFilter+<GreenPipes-IFilter<MassTransit-AzureServiceBusTransport-ClientContext>-Send>d__7.MoveNext (MassTransit.AzureServiceBusTransport, Version=5.1.0.1516, Culture=neutral, PublicKeyToken=b8e0e9f2f1e657fa)
   at GreenPipes.Agents.PipeContextSupervisor`1+<GreenPipes-IPipeContextSource<TContext>-Send>d__8.MoveNext (GreenPipes, Version=2.1.0.106, Culture=neutral, PublicKeyToken=b800c4cfcdeea87b)
   at GreenPipes.Agents.PipeContextSupervisor`1+<GreenPipes-IPipeContextSource<TContext>-Send>d__8.MoveNext (GreenPipes, Version=2.1.0.106, Culture=neutral, PublicKeyToken=b800c4cfcdeea87b)
   at GreenPipes.Agents.PipeContextSupervisor`1+<GreenPipes-IPipeContextSource<TContext>-Send>d__8.MoveNext (GreenPipes, Version=2.1.0.106, Culture=neutral, PublicKeyToken=b800c4cfcdeea87b)
   at MassTransit.AzureServiceBusTransport.Transport.ReceiveTransport+<>c__DisplayClass16_0+<<Receiver>b__0>d.MoveNext (MassTransit.AzureServiceBusTransport, Version=5.1.0.1516, Culture=neutral, PublicKeyToken=b8e0e9f2f1e657fa)
   at MassTransit.AzureServiceBusTransport.Transport.ReceiveTransport+<>c__DisplayClass16_0+<<Receiver>b__0>d.MoveNext (MassTransit.AzureServiceBusTransport, Version=5.1.0.1516, Culture=neutral, PublicKeyToken=b8e0e9f2f1e657fa)
   at MassTransit.Policies.PipeRetryExtensions+<Retry>d__1.MoveNext (MassTransit, Version=5.1.0.1516, Culture=neutral, PublicKeyToken=b8e0e9f2f1e657fa)
   at MassTransit.Policies.PipeRetryExtensions+<Retry>d__1.MoveNext (MassTransit, Version=5.1.0.1516, Culture=neutral, PublicKeyToken=b8e0e9f2f1e657fa)
   at MassTransit.AzureServiceBusTransport.Transport.ReceiveTransport+<Receiver>d__16.MoveNext (MassTransit.AzureServiceBusTransport, Version=5.1.0.1516, Culture=neutral, PublicKeyToken=b8e0e9f2f1e657fa)

Bus Configuration

private IBusControl CreateBusUsingAzureServiceBus()
{
    return Bus.Factory.CreateUsingAzureServiceBus(c =>
    {
        // Normalize the URI
        var uri = Configuration.HostUri;
            uri = ServiceBusEnvironment.CreateServiceUri(AzureServiceBusScheme, uri.Host, "");

        // Configure connection to Azure Service Bus
        var host = c.Host(uri, h =>
        {
            h.SharedAccessSignature(s =>
            {
                s.KeyName         = Configuration.Secret?.UserName;
                s.SharedAccessKey = Configuration.Secret?.Password;
                s.TokenTimeToLive = TimeSpan.FromDays(1);
                s.TokenScope      = TokenScope.Namespace;
            });
        });

        // Disable retries by default
        c.UseRetry(a => a.None());

        // Everything below is for message reception
        if (!Configuration.IsReceiveEnabled)
            return;

        // Ensure long-running consumers will not lose their lock on the message
        c.UseRenewLock();

        // Configure the request queue (shared among all instances, persistent)
        c.ReceiveEndpoint(host, Configuration.QueueName, r =>
        {
            ConfigureFilters(r);
            LoadRequestConsumers(r);
        });

        // Configure the event queue (per-instance, auto-deleted)
        c.ReceiveEndpoint(host, r =>
        {
            // NOTE: Though the queue will auto-delete after 5 min, the
            // subcriptions for it will remain, and worse, will fill up
            // with messages. An external maintenance script should run
            // periodically to clean up those go-nowhere subscriptions.
            // Source: https://github.com/MassTransit/MassTransit/issues/553
            ConfigureFilters(r);
            LoadEventConsumers(r);
        });

        // If the normal plumbing fails, MassTransit will move messages
        // to an error queue.  We currently do not monitor the error
        // queue, so we just need to prevent it from filling up.
        c.ReceiveEndpoint(host, Configuration.QueueName + "_error", r =>
        {
            // Log any request in the error queue
            r.Consumer<LoggingConsumer>();
        });
    });
}

private static void ConfigureFilters(IPipeConfigurator<ConsumeContext> c)
{
    c.UseFilter(new LoggingFilter            <ConsumeContext>());
    c.UseFilter(new ApplicationInsightsFilter<ConsumeContext>());
}

protected virtual void LoadRequestConsumers(IReceiveEndpointConfigurator r)
{
    // Messages that use competing-consumer
    r.LoadConsumersFrom(_context, IsRequestMessage);
}

protected virtual void LoadEventConsumers(IReceiveEndpointConfigurator r)
{
    // Messages that are broadcast
    r.LoadConsumersFrom(_context, IsEventMessage);
}

[phatboyg]

Hopefully not, this wouldn't be a good thing to see this occurring again.

[djmnz]

We are having the same issue in a very similar environment with MassTransit version 5.1.2

[teosangio79]

I have the same issue on Azure Web jobs. Mass transit version 5.1.3

My log is:

[07/02/2018 07:00:24 > 31fe78: INFO] ERROR: ReceiveTransport Faulted: sb://constoso.servicebus.windows.net/RD0003FFDB3E7E_MyBusService_bus_839yyy895cyygr1xbdk77ck9gb?express=true&autodelete=300
[07/02/2018 07:00:24 > 31fe78: INFO] System.InvalidOperationException: The method 'OnMessage' or 'OnMessageAsync' has already been called.
[07/02/2018 07:00:24 > 31fe78: INFO]    at Microsoft.ServiceBus.Messaging.MessageReceiver.OnMessage(MessageReceivePump pump)
[07/02/2018 07:00:24 > 31fe78: INFO]    at Microsoft.ServiceBus.Messaging.QueueClient.OnMessageAsync(Func`2 callback, OnMessageOptions onMessageOptions)
[07/02/2018 07:00:24 > 31fe78: INFO]    at MassTransit.AzureServiceBusTransport.Contexts.QueueClientContext.OnMessageAsync(Func`2 callback, EventHandler`1 exceptionHandler)
[07/02/2018 07:00:24 > 31fe78: INFO]    at MassTransit.AzureServiceBusTransport.Contexts.SharedClientContext.MassTransit.AzureServiceBusTransport.ClientContext.OnMessageAsync(Func`2 callback, EventHandler`1 exceptionHandler)
[07/02/2018 07:00:24 > 31fe78: INFO]    at MassTransit.AzureServiceBusTransport.Transport.Receiver.Start()
[07/02/2018 07:00:24 > 31fe78: INFO]    at MassTransit.AzureServiceBusTransport.Pipeline.MessageReceiverFilter.<GreenPipes-IFilter<MassTransit-AzureServiceBusTransport-ClientContext>-Send>d__7.MoveNext()
[07/02/2018 07:00:24 > 31fe78: INFO] --- End of stack trace from previous location where exception was thrown ---
[07/02/2018 07:00:24 > 31fe78: INFO]    at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
[07/02/2018 07:00:24 > 31fe78: INFO]    at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
[07/02/2018 07:00:24 > 31fe78: INFO]    at GreenPipes.Agents.PipeContextSupervisor`1.<GreenPipes-IPipeContextSource<TContext>-Send>d__8.MoveNext()
[07/02/2018 07:00:24 > 31fe78: INFO] --- End of stack trace from previous location where exception was thrown ---
[07/02/2018 07:00:24 > 31fe78: INFO]    at GreenPipes.Agents.PipeContextSupervisor`1.<GreenPipes-IPipeContextSource<TContext>-Send>d__8.MoveNext()
[07/02/2018 07:00:24 > 31fe78: INFO] --- End of stack trace from previous location where exception was thrown ---
[07/02/2018 07:00:24 > 31fe78: INFO]    at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
[07/02/2018 07:00:24 > 31fe78: INFO]    at GreenPipes.Agents.PipeContextSupervisor`1.<GreenPipes-IPipeContextSource<TContext>-Send>d__8.MoveNext()
[07/02/2018 07:00:24 > 31fe78: INFO] --- End of stack trace from previous location where exception was thrown ---
[07/02/2018 07:00:24 > 31fe78: INFO]    at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
[07/02/2018 07:00:24 > 31fe78: INFO]    at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
[07/02/2018 07:00:24 > 31fe78: INFO]    at MassTransit.AzureServiceBusTransport.Transport.ReceiveTransport.<>c__DisplayClass16_0.<<Receiver>b__0>d.MoveNext()

[phatboyg]

I'm fairly certain this commit will fix it, since I reproduced it by deleting the queue underneath and saw it properly reconnect and reapply the topology to create the queue, and a new QueueClient.

[DraKOCDS]

Are you gonna release this fix with a new nuget package version?

[phatboyg]

Soon, next day or two - been a busy week.

[sharpjs]

Good to hear. We deployed 5.1.2 and 5.1.3 and still see the problem.

[phatboyg]

5.1.4 is now building, and should deploy shortly.

[phatboyg]

This was released, if you still see the issue let me know, otherwise for now I'm going to close this issue.

[teosangio79]

I updated my solution. At the moment it is running well. Let's see in the next days.

Br
Matteo

[sharpjs]

I expect we'll deploy this update today or tomorrow. We'll know if it fixes the issue in a few days.

[phatboyg]

Cool, I was able to reproduce the issue, so I'm hopeful it's resolved.

[kailashravuri]

Is anybody still having this issue? We thought of deploying 5.1.2/5.1.3 to production but @sharpjs methoned he reproduced again in those versions. @sharpjs, @teosangio79 or anyone can confirm this issue in not re-produced in 5.1.4 ?

[teosangio79]

Hi, to me the issue is solved. My message consumer is running for 15 days without any issue at the moment.

[phatboyg]

I haven’t had it reported with 5.1.4, the commit addressing this was in that release.

[sharpjs]

Long-term report: it remains solved for us. cc @kailashravuri