Messages are not received after edgeHub restart - AMQP protocol

[Hammatt]

• OS, version, SKU and CPU architecture used: Ubuntu 18.04 x64
• Application's .NET Target Framework : netcoreapp2.1
• Device: Embedded PC
• SDK version used: 1.17.1

Description of the issue:

Edge modules do not reconnect to the Hub properly after the hub restarts some time during operation. They seem to be able to

Code sample exhibiting the issue:

Any set of two modules which communicate with eachother via AMQP (have not tested other protocols) Here's a sample one I often use:

namespace MessageSender
{
    using System;
    using System.IO;
    using System.Runtime.InteropServices;
    using System.Runtime.Loader;
    using System.Security.Cryptography.X509Certificates;
    using System.Text;
    using System.Threading;
    using System.Threading.Tasks;
    using Microsoft.Azure.Devices.Client;
    using Microsoft.Azure.Devices.Client.Transport.Mqtt;

    class Program
    {
        static int counter;

        static ModuleClient ioTHubModuleClient;

        static void Main(string[] args)
        {
            Init().Wait();

            Task.Run(async () =>
                    {
                        var rnd = new Random();
                        var myMessageSize = 262143-160;
                        while (true)
                        {
                            Console.WriteLine("Generating message...");
                            await Task.Delay(2000);
                            //generate new huge message
                            var myMessageBytes = new byte[myMessageSize];
//                            myMessageSize -= 100;
                            rnd.NextBytes(myMessageBytes);
                            var myMessage = new Message(myMessageBytes);

                            await ioTHubModuleClient.SendEventAsync("input1", myMessage);
                            Console.WriteLine("Message Sent.");
                        }
                    });

            // Wait until the app unloads or is cancelled
            var cts = new CancellationTokenSource();
            AssemblyLoadContext.Default.Unloading += (ctx) => cts.Cancel();
            Console.CancelKeyPress += (sender, cpe) => cts.Cancel();
            WhenCancelled(cts.Token).Wait();
        }

        /// <summary>
        /// Handles cleanup operations when app is cancelled or unloads
        /// </summary>
        public static Task WhenCancelled(CancellationToken cancellationToken)
        {
            var tcs = new TaskCompletionSource<bool>();
            cancellationToken.Register(s => ((TaskCompletionSource<bool>)s).SetResult(true), tcs);
            return tcs.Task;
        }

        /// <summary>
        /// Initializes the ModuleClient and sets up the callback to receive
        /// messages containing temperature information
        /// </summary>
        static async Task Init()
        {
            var amqpSetting = new AmqpTransportSettings(TransportType.Amqp_Tcp_Only);
            ITransportSettings[] settings = { amqpSetting };

            // Open a connection to the Edge runtime
            ioTHubModuleClient = await ModuleClient.CreateFromEnvironmentAsync(settings);
            await ioTHubModuleClient.OpenAsync();
            Console.WriteLine("IoT Hub module client initialized.");

            // Register callback to be called when a message is received by the module
            await ioTHubModuleClient.SetInputMessageHandlerAsync("input1", PipeMessage, ioTHubModuleClient);

            await ioTHubModuleClient.SetDesiredPropertyUpdateCallbackAsync(async (inProperties, inContext) => {Console.WriteLine("Desired Property Update");}, ioTHubModuleClient).ConfigureAwait(false);
        }

        /// <summary>
        /// This method is called whenever the module is sent a message from the EdgeHub.
        /// It just pipe the messages without any change.
        /// It prints all the incoming messages.
        /// </summary>
        static async Task<MessageResponse> PipeMessage(Message message, object userContext)
        {
            Console.WriteLine("Message received.");
            int counterValue = Interlocked.Increment(ref counter);

            var moduleClient = userContext as ModuleClient;
            if (moduleClient == null)
            {
                throw new InvalidOperationException("UserContext doesn't contain " + "expected values");
            }

            Console.WriteLine($"Received message: {counterValue}, size: {message.BodyStream.Length}");

            byte[] messageBytes = message.GetBytes();

            Console.WriteLine($"Read in {messageBytes.Length} bytes");

            return MessageResponse.Completed;
        }
    }
}

Hook two instances of that up to itself.

Console log of the issue:

run docker restart edgeHub to force the edgeHub to restart..

See Azure/iotedge#65 for more info.

[CIPop]

Thanks for reporting @Hammatt seems related to #558 but for AMQP.

[CIPop]

Also related (or duplicate of) #239.

[CIPop]

@Hammatt @abhipsaMisra and I are still investigating both our connection recovery tests and adding more logs.
@varunpuranik has potentially identified the issue manifesting in the Edge scenario and he is working on a fix.

[Hammatt]

Thanks for the update! I can confirm that we have still been seeing this issue occasionally.

[CIPop]

@Hammatt We believe we've identified the cause of this issue: @varunpuranik's PR #611 was tested against both MQTT and AMQP. Once the changes are in, we'll prepare a new release.

[WilliamBerryiii]

@CIPop - any updates on this one?

[CIPop]

@WilliamBerryiii This is a lot more complicated than first expected.

1. @varunpuranik 's PR is breaking the method E2E fault injection tests (i.e. disconnect recovery) and possibly twin. Since we've found this fundamental issue with our SDK, we're blocking all PRs until we get all our E2E recovery tests enabled (I inherited them disabled...)
2. I've added some more logging and was able to at least make our E2E for message reconnect work and they are quite stable. I wasn't able to enable command or twin update (individual tests seem to pass but when all tests run at the same time they fail which is indicative of another issue, potentially with objects kept around after a disconnect/timeout which could lead to 2 or more clients trying to connect with the same identity).
3. I've also found that our E2E fault injection for AMQP are not actually dropping the TCP connection (wrong impl. on the service side) so even with these tests enabled, we're not testing your scenario. I'm looking at other ways to simulate a real TCP connection drop and we'll use Edge as a repro (the problem being that this scenario isn't consistently reproducing according to @varunpuranik).
4. AMQP fixed the following issue which could be related: Bugfix: Message disposition hangs infinitely when the operation times out azure-amqp#123

[WilliamBerryiii]

I'd also add that we should probably find a way to allow community members to run the E2E tests before PRs ... even via a light weight stubbing mechanism - assuming that the CI/CD on a PR triggers a full E2E run on a MSFT owned IoT Hub. (Just thinking about the ease of accepting non-MSFT contributions.)

[CIPop]

@WilliamBerryiii Good point! I meant to document this for a while: Please open a Support Request indicating your test subscription asking to enable Fault Injection.

[sebader]

Would switching modules to MQTT be a workaround or would that not solve it or even bring other issues?

[varunpuranik]

I validated that with the latest release of the SDK - v1.19.0 this issue is fixed - https://www.nuget.org/packages/Microsoft.Azure.Devices.Client/1.19.0

@Hammatt - Can you please validate and if it works, close the issue?
Thanks.

[Hammatt]

Thanks, I will validate this over the next few days and get back to you as soon as I can confirm.

[Hammatt]

Sorry for the delay. I've finally had time to go through a bunch of testing and I haven't been able to reproduce the problem. Thanks!

[az-iot-builder-01]

@CIPop, @Hammatt, @WilliamBerryiii, @sebader, thank you for your contribution to our open-sourced project! Please help us improve by filling out this 2-minute customer satisfaction survey