edgeHub: resume store-and-forward promptly after reconnect

edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.CloudProxy/ConnectivityAwareClient.cs

@@ -27,6 +27,17 @@ class ConnectivityAwareClient : IClient
         readonly IIdentity identity;
         ConnectionStatusChangesHandler connectionStatusChangedHandler;
 
+        // The SDK callback can emit rapid alternating status changes. Debounce it so
+        // edgeHub reacts quickly to a stable status without propagating flap churn.
+        static readonly TimeSpan DebounceWindow = TimeSpan.FromSeconds(2);
+        readonly object debounceLock = new object();
+        Timer debounceTimer;
+        long debounceGeneration;
+        bool isOpen;
+
+        // Enabled for the draft E2E experiment. Make this configurable before merge.
+        bool sdkBridgeEnabled = true;
+
         public ConnectivityAwareClient(IClient client, IDeviceConnectivityManager deviceConnectivityManager, IIdentity identity)
         {
             this.identity = Preconditions.CheckNotNull(identity, nameof(identity));
@@ -38,6 +49,7 @@ public ConnectivityAwareClient(IClient client, IDeviceConnectivityManager device
 
         public async Task CloseAsync()
         {
+            this.CancelDebouncedSdkBridge();
             await this.underlyingClient.CloseAsync();
             this.isConnected.Set(false);
             this.deviceConnectivityManager.DeviceConnected -= this.HandleDeviceConnectedEvent;
@@ -60,6 +72,11 @@ public Task RejectAsync(string messageId) =>
         public async Task OpenAsync()
         {
             await this.InvokeFunc(() => this.underlyingClient.OpenAsync(), nameof(this.OpenAsync));
+            lock (this.debounceLock)
+            {
+                this.isOpen = true;
+            }
+
             this.deviceConnectivityManager.DeviceConnected += this.HandleDeviceConnectedEvent;
             this.deviceConnectivityManager.DeviceDisconnected += this.HandleDeviceDisconnectedEvent;
         }
@@ -92,6 +109,7 @@ public Task UpdateReportedPropertiesAsync(TwinCollection reportedProperties) =>
 
         public void Dispose()
         {
+            this.CancelDebouncedSdkBridge();
             this.deviceConnectivityManager.DeviceConnected -= this.HandleDeviceConnectedEvent;
             this.deviceConnectivityManager.DeviceDisconnected -= this.HandleDeviceDisconnectedEvent;
             this.isConnected.Set(false);
@@ -123,20 +141,100 @@ void HandleDeviceDisconnectedEvent()
         void InternalConnectionStatusChangedHandler(ConnectionStatus status, ConnectionStatusChangeReason reason)
         {
             Events.ReceivedDeviceSdkCallback(this.identity, status, reason);
-            // @TODO: Ignore callback from Device SDK since it seems to be generating a lot of spurious Connected/NotConnected callbacks
-            /*
-            if (status == ConnectionStatus.Connected)
+
+            if (!this.sdkBridgeEnabled ||
+                (status != ConnectionStatus.Connected &&
+                 status != ConnectionStatus.Disconnected &&
+                 status != ConnectionStatus.Disconnected_Retrying &&
+                 status != ConnectionStatus.Disabled))
+            {
+                return;
+            }
+
+            lock (this.debounceLock)
+            {
+                if (!this.isOpen)
+                {
+                    return;
+                }
+
+                long generation = ++this.debounceGeneration;
+                this.debounceTimer?.Dispose();
+                this.debounceTimer = new Timer(
+                    _ => this.OnDebounceElapsed(generation, status),
+                    null,
+                    DebounceWindow,
+                    Timeout.InfiniteTimeSpan);
+            }
+        }
+
+        void OnDebounceElapsed(long generation, ConnectionStatus status) =>
+            _ = this.ApplyDebouncedStatusAsync(generation, status);
+
+        async Task ApplyDebouncedStatusAsync(long generation, ConnectionStatus status)
+        {
+            lock (this.debounceLock)
             {
-                this.deviceConnectivityManager.CallSucceeded();
-                this.HandleDeviceConnectedEvent();
+                if (!this.isOpen || generation != this.debounceGeneration)
+                {
+                    return;
+                }
+
+                bool alreadyEffective = status == ConnectionStatus.Connected
+                    ? this.isConnected.Get()
+                    : !this.isConnected.Get();
+                if (alreadyEffective)
+                {
+                    return;
+                }
+
+                this.debounceTimer?.Dispose();
+                this.debounceTimer = null;
             }
-            else if (status == ConnectionStatus.Disconnected || status == ConnectionStatus.Disabled)
+
+            try
+            {
+                if (status == ConnectionStatus.Connected)
+                {
+                    await this.deviceConnectivityManager.CallSucceeded();
+                }
+                else
+                {
+                    await this.deviceConnectivityManager.CallTimedOut();
+                }
+
+                lock (this.debounceLock)
+                {
+                    if (!this.isOpen || generation != this.debounceGeneration)
+                    {
+                        return;
+                    }
+                }
+
+                if (status == ConnectionStatus.Connected)
+                {
+                    this.HandleDeviceConnectedEvent();
+                }
+                else
+                {
+                    this.HandleDeviceDisconnectedEvent();
+                }
+            }
+            catch (Exception ex)
+            {
+                Events.OperationFailed(this.identity, "applying debounced SDK connection status", ex);
+            }
+        }
+
+        void CancelDebouncedSdkBridge()
+        {
+            lock (this.debounceLock)
             {
-                this.deviceConnectivityManager.CallTimedOut();
-                this.HandleDeviceDisconnectedEvent();
+                this.isOpen = false;
+                ++this.debounceGeneration;
+                this.debounceTimer?.Dispose();
+                this.debounceTimer = null;
             }
-            this.connectionStatusChangedHandler?.Invoke(status, reason);
-            */
         }
 
         async Task<T> InvokeFunc<T>(Func<Task<T>> func, string operation, bool useForConnectivityCheck = true)

edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.CloudProxy/DeviceConnectivityManager.cs

@@ -87,6 +87,8 @@ public DeviceConnectivityManager(
 
         public event EventHandler DeviceDisconnected;
 
+        public event EventHandler ConnectivityRecovered;
+
         enum State
         {
             Connected,
@@ -140,6 +142,7 @@ void ResetConnectedTimer()
         void OnConnected()
         {
             Events.OnConnected();
+            this.ConnectivityRecovered?.Invoke(this, EventArgs.Empty);
             this.connectedTimer.Start();
         }
 

edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Core/IDeviceConnectivityManager.cs

@@ -10,6 +10,10 @@ public interface IDeviceConnectivityManager
 
         event EventHandler DeviceDisconnected;
 
+        // Unlike DeviceConnected, this event also fires when a short outage recovers
+        // from the intermediate Trying state without first reaching Disconnected.
+        event EventHandler ConnectivityRecovered;
+
         Task CallSucceeded();
 
         Task CallTimedOut();

edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Core/NullDeviceConnectivityManager.cs

@@ -25,6 +25,12 @@ public event EventHandler DeviceDisconnected
             remove { }
         }
 
+        public event EventHandler ConnectivityRecovered
+        {
+            add { }
+            remove { }
+        }
+
         public Task CallSucceeded() => Task.CompletedTask;
 
         public Task CallTimedOut() => Task.CompletedTask;

edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Service/modules/RoutingModule.cs

@@ -430,7 +430,16 @@ protected override void Load(ContainerBuilder builder)
                         {
                             var endpointExecutorConfig = c.Resolve<EndpointExecutorConfig>();
                             var messageStore = await c.Resolve<Task<IMessageStore>>();
-                            IEndpointExecutorFactory endpointExecutorFactory = new StoringAsyncEndpointExecutorFactory(endpointExecutorConfig, new AsyncEndpointExecutorOptions(10, TimeSpan.FromSeconds(10)), messageStore);
+                            // A recovered connection can land in the middle of an endpoint retry
+                            // backoff (up to 60 seconds). Wake parked FSMs instead of waiting out
+                            // the remaining backoff before store-and-forward resumes.
+                            var retrySignal = new EndpointExecutorRetrySignal();
+                            c.Resolve<IDeviceConnectivityManager>().ConnectivityRecovered += (_, __) => retrySignal.RequestRetry();
+                            IEndpointExecutorFactory endpointExecutorFactory = new StoringAsyncEndpointExecutorFactory(
+                                endpointExecutorConfig,
+                                new AsyncEndpointExecutorOptions(10, TimeSpan.FromSeconds(10)),
+                                messageStore,
+                                retrySignal);
                             return endpointExecutorFactory;
                         })
                     .As<Task<IEndpointExecutorFactory>>()

edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/IEndpointExecutorRetrySignal.cs

@@ -0,0 +1,18 @@
+// Copyright (c) Microsoft. All rights reserved.
+namespace Microsoft.Azure.Devices.Routing.Core.Endpoints
+{
+    using System;
+
+    // Decouples a transport connectivity-recovery event from endpoint FSM retries.
+    public interface IEndpointExecutorRetrySignal
+    {
+        event EventHandler RetryRequested;
+    }
+
+    public sealed class EndpointExecutorRetrySignal : IEndpointExecutorRetrySignal
+    {
+        public event EventHandler RetryRequested;
+
+        public void RequestRetry() => this.RetryRequested?.Invoke(this, EventArgs.Empty);
+    }
+}

edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/StoringAsyncEndpointExecutor.cs

@@ -30,6 +30,7 @@ public class StoringAsyncEndpointExecutor : IEndpointExecutor
         readonly CancellationTokenSource cts = new CancellationTokenSource();
         readonly ICheckpointerFactory checkpointerFactory;
         readonly EndpointExecutorConfig config;
+        readonly IEndpointExecutorRetrySignal retrySignal;
         AtomicReference<ImmutableDictionary<uint, EndpointExecutorFsm>> prioritiesToFsms;
         EndpointExecutorFsm lastUsedFsm;
 
@@ -38,15 +39,21 @@ public StoringAsyncEndpointExecutor(
             ICheckpointerFactory checkpointerFactory,
             EndpointExecutorConfig config,
             AsyncEndpointExecutorOptions options,
-            IMessageStore messageStore)
+            IMessageStore messageStore,
+            IEndpointExecutorRetrySignal retrySignal = null)
         {
             this.Endpoint = Preconditions.CheckNotNull(endpoint);
             this.checkpointerFactory = Preconditions.CheckNotNull(checkpointerFactory);
             this.config = Preconditions.CheckNotNull(config);
             this.options = Preconditions.CheckNotNull(options);
             this.messageStore = messageStore;
+            this.retrySignal = retrySignal;
             this.sendMessageTask = Task.Run(this.SendMessagesPump);
             this.prioritiesToFsms = new AtomicReference<ImmutableDictionary<uint, EndpointExecutorFsm>>(ImmutableDictionary<uint, EndpointExecutorFsm>.Empty);
+            if (this.retrySignal != null)
+            {
+                this.retrySignal.RetryRequested += this.HandleRetryRequested;
+            }
         }
 
         public Endpoint Endpoint { get; }
@@ -90,6 +97,11 @@ public async Task CloseAsync()
             {
                 if (!this.closed.GetAndSet(true))
                 {
+                    if (this.retrySignal != null)
+                    {
+                        this.retrySignal.RetryRequested -= this.HandleRetryRequested;
+                    }
+
                     this.cts.Cancel();
                     // Require to close all FSMs to complete currently executing command if any in order to unblock sendMessageTask.
                     ImmutableDictionary<uint, EndpointExecutorFsm> snapshot = this.prioritiesToFsms;
@@ -299,10 +311,37 @@ async Task ProcessMessages(IMessage[] messages, EndpointExecutorFsm fsm)
             await command.Completion;
         }
 
+        void HandleRetryRequested(object sender, EventArgs eventArgs)
+        {
+            if (!this.closed)
+            {
+                _ = Task.Run(this.RetryFailingEndpointsAsync);
+            }
+        }
+
+        async Task RetryFailingEndpointsAsync()
+        {
+            try
+            {
+                ImmutableDictionary<uint, EndpointExecutorFsm> snapshot = this.prioritiesToFsms;
+                await Task.WhenAll(snapshot.Values.Select(fsm => fsm.RetryNowAsync()));
+                Events.RetryRequested(this);
+            }
+            catch (Exception ex)
+            {
+                Events.RetryRequestFailure(this, ex);
+            }
+        }
+
         void Dispose(bool disposing)
         {
             if (disposing)
             {
+                if (this.retrySignal != null)
+                {
+                    this.retrySignal.RetryRequested -= this.HandleRetryRequested;
+                }
+
                 this.cts.Dispose();
                 ImmutableDictionary<uint, EndpointExecutorFsm> snapshot = this.prioritiesToFsms;
                 this.prioritiesToFsms.CompareAndSet(snapshot, ImmutableDictionary<uint, EndpointExecutorFsm>.Empty);
@@ -390,7 +429,9 @@ enum EventIds
                 Close,
                 CloseSuccess,
                 CloseFailure,
-                ErrorInPopulatePump
+                ErrorInPopulatePump,
+                RetryRequested,
+                RetryRequestFailure
             }
 
             public static void AddMessageSuccess(StoringAsyncEndpointExecutor executor, long offset, uint priority, uint timeToLiveSecs)
@@ -483,6 +524,16 @@ public static void ErrorInPopulatePump(Exception ex)
             {
                 Log.LogWarning((int)EventIds.ErrorInPopulatePump, ex, "Error in populate messages pump");
             }
+
+            public static void RetryRequested(StoringAsyncEndpointExecutor executor)
+            {
+                Log.LogDebug((int)EventIds.RetryRequested, "[RetryRequested] Retried failing endpoint FSMs immediately for EndpointId: {0}.", executor.Endpoint.Id);
+            }
+
+            public static void RetryRequestFailure(StoringAsyncEndpointExecutor executor, Exception ex)
+            {
+                Log.LogWarning((int)EventIds.RetryRequestFailure, ex, "[RetryRequestFailure] Failed to retry endpoint FSMs for EndpointId: {0}.", executor.Endpoint.Id);
+            }
         }
 
         static class MetricsV0

edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/StoringAsyncEndpointExecutorFactory.cs

@@ -11,12 +11,18 @@ public class StoringAsyncEndpointExecutorFactory : IEndpointExecutorFactory
         readonly EndpointExecutorConfig config;
         readonly AsyncEndpointExecutorOptions options;
         readonly IMessageStore messageStore;
+        readonly IEndpointExecutorRetrySignal retrySignal;
 
-        public StoringAsyncEndpointExecutorFactory(EndpointExecutorConfig config, AsyncEndpointExecutorOptions options, IMessageStore messageStore)
+        public StoringAsyncEndpointExecutorFactory(
+            EndpointExecutorConfig config,
+            AsyncEndpointExecutorOptions options,
+            IMessageStore messageStore,
+            IEndpointExecutorRetrySignal retrySignal = null)
         {
             this.config = Preconditions.CheckNotNull(config, nameof(config));
             this.options = Preconditions.CheckNotNull(options, nameof(options));
             this.messageStore = Preconditions.CheckNotNull(messageStore, nameof(messageStore));
+            this.retrySignal = retrySignal;
         }
 
         public Task<IEndpointExecutor> CreateAsync(Endpoint endpoint, IList<uint> priorities) => this.CreateAsync(endpoint, priorities, new NullCheckpointerFactory(), this.config);
@@ -29,7 +35,7 @@ public async Task<IEndpointExecutor> CreateAsync(Endpoint endpoint, IList<uint>
             Preconditions.CheckNotNull(checkpointerFactory, nameof(checkpointerFactory));
             Preconditions.CheckNotNull(endpointExecutorConfig, nameof(endpointExecutorConfig));
 
-            var endpointExecutor = new StoringAsyncEndpointExecutor(endpoint, checkpointerFactory, endpointExecutorConfig, this.options, this.messageStore);
+            var endpointExecutor = new StoringAsyncEndpointExecutor(endpoint, checkpointerFactory, endpointExecutorConfig, this.options, this.messageStore, this.retrySignal);
             await endpointExecutor.UpdatePriorities(priorities, Option.None<Endpoint>());
             return endpointExecutor;
         }