Azure-iot-sdk-csharp: MqttIotHubAdapter can throw unobserved NullReferenceException when reconnecting

[ccruden-shookiot]

Context

• OS, version, SKU and CPU architecture used: Windows Server 2016 x64
• Application's .NET Target Framework : .NET framework 4.8
• Device: Cloud hosted VM
• SDK version used: Microsoft.Azure.Devices.Client 1.31.1

Description of the issue

Our application is a service hosted on a cloud VM that continually writes data to an MQTT IOT hub. We use an SAS token to authenticate the connection, so the connection is dropped and refreshed every 65 minutes. It appears that when that refresh happens, if something goes wrong, MqttIotHubAdapter.ShutdownOnError (now ShutdownOnErrorAsync) can handle an exception which results in a NullReferenceException being thrown on a task / thread which is unobserved, and so bubbles up to the top level and causes the application itself to exit.

Event viewer provided the following information:
Framework Version: v4.0.30319
Description: The process was terminated due to an unhandled exception.
Exception Info: System.NullReferenceException
at Microsoft.Azure.Devices.Client.Transport.Mqtt.OrderedTwoPhaseWorkQueue`2[[System.Int32, mscorlib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089],[System.__Canon, mscorlib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089]].Abort(System.Exception)
at Microsoft.Azure.Devices.Client.Transport.Mqtt.MqttIotHubAdapter+d__69.MoveNext()
at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
at System.Threading.ExecutionContext.RunInternal(System.Threading.ExecutionContext, System.Threading.ContextCallback, System.Object, Boolean)
at System.Threading.ExecutionContext.Run(System.Threading.ExecutionContext, System.Threading.ContextCallback, System.Object, Boolean)
at System.Threading.QueueUserWorkItemCallback.System.Threading.IThreadPoolWorkItem.ExecuteWorkItem()
at System.Threading.ThreadPoolWorkQueue.Dispatch()

At the same time, the IOT hub's AzureDiagnostics log shows a deviceDisconnect with result DeviceConnectionClosedRemotely and a deviceConnect with result IotHubUnauthorized. Normally this would be followed by a deviceConnect, but in this case it wasn't.

Code sample exhibiting the issue

`

class IOTHubConsumer: IDisposable
{
    private DeviceClient _IOTHubClient;
    private ILogger _log;

    public IOTHubConsumer(string connString, ILogger log)
    {
        _connString = connString;
        _log = log;

        try
        {
            _log.LogDebug($"IOTHubConsumer {_type} created with string: {connString}");
            _IOTHubClient = DeviceClient.CreateFromConnectionString(connString, TransportType.Mqtt);
            this.connectToHub();
        }
        catch (Exception ex)
        {
            _log.LogError(ex.Message);
            throw ex;
        }
    }
    private void connectToHub()
    {
        Task openTask = _IOTHubClient.OpenAsync();
        try
        {
            openTask.Wait();
            _log.LogDebug("Successfully Opened IOTHub");
        }
        catch (Exception ex)
        {
            _log.LogError($"IOTHub Open: {ex.Message}");
            throw ex;
        }
    }
    public Task ConsumeAsync(List<JsonMessage> list)
    {
        return Task.Factory.StartNew(() =>
        {
            try
            {
                //filter out null values generated potentially by poor decoded data
                List<JsonMessage> filtered = (from c in list where c != null select c).ToList();
                //handle suspect batch items
                List<JsonMessage> suspect = (from c in filtered where c.IsSuspect == true select c).ToList();
                if (suspect.Any())
                {
                    JsonWrapper suspectWrapper = new JsonWrapper();
                    suspectWrapper.PiServer = _serverName;
                    suspectWrapper.DataState = "Suspect";
                    suspectWrapper.events = suspect;
                    string suspectBatchString = JsonConvert.SerializeObject(suspectWrapper, _jsonSettings);
                    Message msg = new Message(Encoding.UTF8.GetBytes(suspectBatchString))
                    {
                        ContentType = "application/json",
                        ContentEncoding = Encoding.UTF8.ToString()
                    };
                    int length = msg.GetBytes().Length;

                    Task sendTask = _IOTHubClient.SendEventAsync(msg); ;
                    try
                    {
                        sendTask.Wait();
                        _log.LogDebug($"Sent a Suspect batch To IOTHub: {length}");

                    }
                    catch (Exception ex)
                    {
                        _log.LogDebug("Error encountered while sending Suspect data to IoTHub {" + ex.Message + "}");
                        if (ex.InnerException != null)
                        {
                            _log.LogDebug($"Inner Exception: {ex.InnerException.Message}");
                            _log.LogDebug($"Base Exception: {ex.GetBaseException().Message}");
                            _log.LogDebug($"Base Exception StackTrace: {ex.GetBaseException().StackTrace}");
                        }
                    }
                }
                //handle normal batch items
                List<JsonMessage> normal = (from c in filtered where c.IsSuspect == false select c).ToList();
                if (normal.Any())
                {
                    JsonWrapper wrapper = new JsonWrapper();
                    wrapper.PiServer = _serverName;
                    wrapper.DataState = "Normal";
                    wrapper.events = normal;
                    string batchString = JsonConvert.SerializeObject(wrapper, _jsonSettings);
                    Message msg = new Message(Encoding.UTF8.GetBytes(batchString))
                    {
                        ContentType = "application/json",
                        ContentEncoding = Encoding.UTF8.ToString()
                    };
                    int length = msg.GetBytes().Length;

                    Task sendTask = _IOTHubClient.SendEventAsync(msg); ;
                    try
                    {
                        sendTask.Wait();
                        _log.LogDebug($"Sent a batch To IOTHub: {length}");

                    }
                    catch (Exception ex)
                    {
                        _log.LogDebug("Error encountered while sending data to IoTHub {" + ex.Message + "}");
                        if (ex.InnerException != null)
                        {
                            _log.LogDebug($"Inner Exception: {ex.InnerException.Message}");
                            _log.LogDebug($"Base Exception: {ex.GetBaseException().Message}");
                            _log.LogDebug($"Base Exception StackTrace: {ex.GetBaseException().StackTrace}");
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                _log.LogError(ex, "IOTHubConsumer.ConsumeAsync");
            }
        });
    }

`

[drwill-ms]

We'll look into the NRE.

Question for you about the code, why is the method running the SDK async methods not awaiting them? Calling .Wait() is not a safe option.

Also, don't forget to dispose objects that are IDisposable, such as Message and DeviceClient.

Finally, if you aren't doing this in your actual application, be sure to hook up to the connection status callback and respond appropriately. You can view our reconnection sample for an illustration of all the needed actions.

[ccruden-shookiot]

No good reason for the .Wait()s - those will get changed now that the code is coming under the microscope.
Message I honestly would not have thought was IDisposable had you not mentioned it. That will get changed. DeviceClient is disposed in the usual way (as a managed member in the default IDisposable pattern).
I'll take a look at the connection status callback - is that typically required? Things did seem to be working correctly without the handler, and as this is a fairly high throughput class/call, I'd like to avoid introducing any extra synchronization code...

[drwill-ms]

@ccruden-shookiot I'm not intending to scrutinize your code in the context of the bug report, but I try to take the opportunity to make sure our customers use good patterns for success. You'd be surprised how often we help eliminate bugs from these interactions. I know you submitted a sample illustrating the bug, and I very much do appreciate that. My apologies if my response came across differently.

For async await, the pattern we try to follow is to start async in Main, and stay async until a sync call is introduced. That means we avoid synchronous methods calling async methods, and when we encounter that we try to make the whole call chain above it async. The biggest challenge is with constructors and properties, which can't be async. I'll admit, our current code base has a handful of places where a sync method calls an async one. We've tried to eliminate them, but it is not always possible. One such case is where we derive from another libraries abstract class and the call signature is determined by them and they call the method. :(

I mention DeviceClient lifetime/disposal and connection status callback because there are cases in disconnection where the SDK will give up retrying, and it is the responsibility of the device app to determine to retry on its own, or not. You can look at the sample I linked to see those cases. You may implement a device app without connection status callback, and depending on the reliability of your connection and a certain amount of luck it may run just fine for many moons. But time will catch up with your app eventually, even if it is a servicing event in IoT Hub where your device gets disconnected and needs to reconnect (to the new stamp). Generally people want their IoT devices apps to continue to run for a long, long time without human intervention (as much as possible), in which case this would be necessary.

Finally, can you give us a small taste of the Main function? How does it call ConsumeAsync?

[ccruden-shookiot]

Code suggestions are always welcome - anything that gets code running better.

ConsumeAsync is called in an ActionBlock at the end of a chain of TPL dataflow blocks. A TransformBlock at the start brings in events from the datasource and transforms them into JsonMessage instances. We use a BatchBlock to group those up, and then the ActionBlock is called using

block = new ActionBlock

[drwill-ms]

The NRE happens in a file we haven't edited in a while. I'm putting out a PR to clean it up a bit and @jamdavi will continue to investigate, as I'll be OOF for a week.

Thanks so much for the bug report and extra info. :)

It sounds like you might be blocked by this bug. Is that right?

[ccruden-shookiot]

This hasn't been a common occurrence - it's the first time we've seen it in a couple months of running this service at several sites sending thousands of messages per second. But it's concerning because it resulted in the service dying in a way that we don't have a nice way to recover from, and the service is meant to have near 100% uptime. (There seems to be TaskScheduler.UnobservedTaskException to hook into, but I have no idea what state that leaves the connection in.)

[ccruden-shookiot]

In terms of being able to fix it ourselves to make sure it doesn't reoccur - yeah, we're blocked.

[ccruden-shookiot]

Interesting side note to this.... We tried adding a handler for unobserved exceptions to the program as a whole:

private static async Task Main(string[] args) {
            TaskScheduler.UnobservedTaskException += (s,e) => 
            {
                e.Exception.Handle(ex => 
                {
                    Log.Logger?.Error(ex, "Unobserved exception"); 
                    return true; 
                });

                e.SetObserved();
            };
...

... just to see what would happen. Surprisingly, it got triggered rather more than we'd expected. For instance, it catches a SocketException with no stack trace about 10m after the regular disconnect/reconnect cycle - that's anonymous enough that it could have come from other client code we use though. When we disabled the IOT device we were sending to and waited, we eventually caught a bunch of Microsoft.Azure.Devices.Client.Exceptions.IotHubCommunicationException exceptions in that handler as well - which, 30s later, generated OperationCanceledExceptions on the SendAsync call.

[jamdavi]

That is interesting. Not sure if it's related to this issue, but I'd like to see the call stacks none the less to be sure and investigate.

As for the NRE that we're seeing, @drwill-ms made these changes in #1738. Can you reach out to me by e-mail (jamdavi at microsoft) and I'll send you a link to a private build with these changes.

[ccruden-shookiot]

I've included an extract from the log here. The first debug message is just us completing a message send to the IOT hub. Not long after that I switched the device at the end of that connection from enabled to disabled, and that triggered the "IOT hub disconnected - retrying" message. (We added a ConnectionStatusChanges handler - that message comes from the status changing to ConnectionStatus.Disconnected_Retrying.) After that, nothing happens while messages queue up waiting to be successfully sent, then we get the unobserved exception message below repeated 8 times.

2021-01-08 17:58:16 [DBG] Sent a batch To IOTHub: 11525
2021-01-08 17:58:25 [DBG] IOT hub disconnected - retrying.
2021-01-08 18:01:17 [ERR] Unobserved exception
Microsoft.Azure.Devices.Client.Exceptions.IotHubCommunicationException: CONNECT failed: RefusedServerUnavailable
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at Microsoft.Azure.Devices.Client.Transport.Mqtt.MqttTransportHandler.d__83.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at Microsoft.Azure.Devices.Client.Transport.Mqtt.MqttTransportHandler.d__63.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at Microsoft.Azure.Devices.Client.Transport.ProtocolRoutingDelegatingHandler.d__7.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at Microsoft.Azure.Devices.Client.Transport.ErrorDelegatingHandler.<>c__DisplayClass26_0.<b__0>d.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at Microsoft.Azure.Devices.Client.Transport.ErrorDelegatingHandler.d__27`1.MoveNext()
--- End of stack trace from previous location where exception was thrown ---
at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
at Microsoft.Azure.Devices.Client.Transport.RetryDelegatingHandler.<>c__DisplayClass39_0.<b__0>d.MoveNext()

[jamdavi]

Keeping thread up to date. Had a small conversation with Charles offline, will update this thread once we decide where to go next.