From b825135841e5e9734d7c3f82cd326207ab5951ce Mon Sep 17 00:00:00 2001 From: John Lian Date: Fri, 10 Jul 2026 10:09:47 -0700 Subject: [PATCH 1/2] edgeHub: debounce Device SDK connection-status callback for fast, flap-safe reconnect detection edgeHub's ConnectivityAwareClient ignored the Device SDK connection-status callback (commented out due to spurious alternating Connected/Disconnected callbacks) and relied solely on DeviceConnectivityManager poll timers (disconnected-check 2 min / connected-check 5 min). A short upstream outage (e.g. a parent edgeHub restart) is therefore invisible until the next poll, which turns a ~15s reconnect into a multi-minute store-and-forward stall downstream (observed in the nested-edge RouteMessageL3LeafToL4Module E2E). Re-enable the SDK callback but debounce it: a status change only takes effect after a stable DebounceWindow, and a generation counter cancels superseded timers. This gives immediate detection on a genuine transition while collapsing flap churn (20 alternating raw edges settle to exactly 1 effective transition). Effective-status dedup avoids redundant manager calls; the timer is cancelled on Close/Dispose to prevent post-close callbacks; manager-call failures are observed and logged. Draft/post-LTS: sdkBridgeEnabled + DebounceWindow are currently hardcoded for the E2E experiment and must be made configuration-driven before merge; needs full [Integration] + a real nested-edge run. Adds DebouncedSdkBridgeTest covering flap settling, generation cancellation, close/dispose cancellation, and effective-status dedup. --- .../ConnectivityAwareClient.cs | 118 ++++++++++++-- .../DebouncedSdkBridgeTest.cs | 152 ++++++++++++++++++ 2 files changed, 260 insertions(+), 10 deletions(-) create mode 100644 edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/DebouncedSdkBridgeTest.cs diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.CloudProxy/ConnectivityAwareClient.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.CloudProxy/ConnectivityAwareClient.cs index f3f55b9bc8e..2b132461fc5 100644 --- a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.CloudProxy/ConnectivityAwareClient.cs +++ b/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 InvokeFunc(Func> func, string operation, bool useForConnectivityCheck = true) diff --git a/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/DebouncedSdkBridgeTest.cs b/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/DebouncedSdkBridgeTest.cs new file mode 100644 index 00000000000..020009810be --- /dev/null +++ b/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/DebouncedSdkBridgeTest.cs @@ -0,0 +1,152 @@ +// Copyright (c) Microsoft. All rights reserved. +namespace Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test +{ + using System; + using System.Diagnostics; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Azure.Devices.Client; + using Microsoft.Azure.Devices.Edge.Hub.Core; + using Microsoft.Azure.Devices.Edge.Hub.Core.Identity; + using Microsoft.Azure.Devices.Edge.Util.Test.Common; + using Moq; + using Xunit; + + [Unit] + public class DebouncedSdkBridgeTest + { + [Fact] + public async Task FlapSettlingConnectedProducesOneFastTransition() + { + var manager = CreateConnectivityManager(); + var underlying = CreateUnderlyingClient(out ConnectionStatusChangesHandler sdkHandler); + using var client = new ConnectivityAwareClient( + underlying.Object, + manager.Object, + Mock.Of(i => i.Id == "d1/$edgeHub")); + + var connected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var disconnected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + client.SetConnectionStatusChangedHandler((status, reason) => + { + if (status == ConnectionStatus.Connected) + { + connected.TrySetResult(DateTime.UtcNow); + } + else + { + disconnected.TrySetResult(DateTime.UtcNow); + } + }); + + await client.OpenAsync(); + + // Establish a disconnected baseline so the final Connected edge is observable. + sdkHandler(ConnectionStatus.Disconnected, ConnectionStatusChangeReason.No_Network); + await disconnected.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + manager.Invocations.Clear(); + connected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + disconnected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + for (int i = 0; i < 10; i++) + { + sdkHandler(ConnectionStatus.Disconnected, ConnectionStatusChangeReason.No_Network); + sdkHandler(ConnectionStatus.Connected, ConnectionStatusChangeReason.Connection_Ok); + await Task.Delay(50); + } + + DateTime finalEdge = DateTime.UtcNow; + sdkHandler(ConnectionStatus.Connected, ConnectionStatusChangeReason.Connection_Ok); + DateTime notification = await connected.Task.WaitAsync(TimeSpan.FromSeconds(5)); + TimeSpan elapsed = notification - finalEdge; + + Assert.InRange(elapsed, TimeSpan.FromSeconds(1.5), TimeSpan.FromSeconds(4)); + Assert.False(disconnected.Task.IsCompleted); + manager.Verify(m => m.CallSucceeded(), Times.Once); + manager.Verify(m => m.CallTimedOut(), Times.Never); + } + + [Fact] + public async Task SustainedDisconnectProducesOneFastTransition() + { + var manager = CreateConnectivityManager(); + var underlying = CreateUnderlyingClient(out ConnectionStatusChangesHandler sdkHandler); + using var client = new ConnectivityAwareClient( + underlying.Object, + manager.Object, + Mock.Of(i => i.Id == "d1/$edgeHub")); + + var disconnected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + client.SetConnectionStatusChangedHandler((status, reason) => + { + if (status != ConnectionStatus.Connected) + { + disconnected.TrySetResult(DateTime.UtcNow); + } + }); + + await client.OpenAsync(); + manager.Invocations.Clear(); + + DateTime edge = DateTime.UtcNow; + sdkHandler(ConnectionStatus.Disconnected, ConnectionStatusChangeReason.No_Network); + DateTime notification = await disconnected.Task.WaitAsync(TimeSpan.FromSeconds(5)); + TimeSpan elapsed = notification - edge; + + Assert.InRange(elapsed, TimeSpan.FromSeconds(1.5), TimeSpan.FromSeconds(4)); + manager.Verify(m => m.CallTimedOut(), Times.Once); + manager.Verify(m => m.CallSucceeded(), Times.Never); + } + + [Fact] + public async Task CloseCancelsPendingTransition() + { + var manager = CreateConnectivityManager(); + var underlying = CreateUnderlyingClient(out ConnectionStatusChangesHandler sdkHandler); + using var client = new ConnectivityAwareClient( + underlying.Object, + manager.Object, + Mock.Of(i => i.Id == "d1/$edgeHub")); + + int disconnected = 0; + client.SetConnectionStatusChangedHandler((status, reason) => + { + if (status != ConnectionStatus.Connected) + { + Interlocked.Increment(ref disconnected); + } + }); + + await client.OpenAsync(); + manager.Invocations.Clear(); + + sdkHandler(ConnectionStatus.Disconnected, ConnectionStatusChangeReason.No_Network); + await client.CloseAsync(); + await Task.Delay(TimeSpan.FromSeconds(3)); + + Assert.Equal(0, disconnected); + manager.Verify(m => m.CallTimedOut(), Times.Never); + } + + static Mock CreateConnectivityManager() + { + var manager = new Mock(); + manager.Setup(m => m.CallSucceeded()).Returns(Task.CompletedTask); + manager.Setup(m => m.CallTimedOut()).Returns(Task.CompletedTask); + return manager; + } + + static Mock CreateUnderlyingClient(out ConnectionStatusChangesHandler sdkHandler) + { + ConnectionStatusChangesHandler capturedHandler = null; + var underlying = new Mock(); + underlying.Setup(c => c.SetConnectionStatusChangedHandler(It.IsAny())) + .Callback(handler => capturedHandler = handler); + underlying.Setup(c => c.OpenAsync()).Returns(Task.CompletedTask); + underlying.Setup(c => c.CloseAsync()).Returns(Task.CompletedTask); + sdkHandler = (status, reason) => capturedHandler(status, reason); + return underlying; + } + } +} From ce1f11487ffd6cb392d725afad24b2a67e09aa8b Mon Sep 17 00:00:00 2001 From: John Lian Date: Sat, 11 Jul 2026 17:34:13 -0700 Subject: [PATCH 2/2] edgeHub: retry queued messages immediately on reconnect --- .../DeviceConnectivityManager.cs | 3 + .../IDeviceConnectivityManager.cs | 4 ++ .../NullDeviceConnectivityManager.cs | 6 ++ .../modules/RoutingModule.cs | 11 +++- .../endpoints/IEndpointExecutorRetrySignal.cs | 18 ++++++ .../endpoints/StoringAsyncEndpointExecutor.cs | 55 ++++++++++++++++++- .../StoringAsyncEndpointExecutorFactory.cs | 10 +++- .../statemachine/EndpointExecutorFsm.cs | 26 +++++++-- .../ConnectivityAwareClientTest.cs | 6 ++ .../DeviceConnectivityManagerTest.cs | 23 ++++++++ .../ConnectionManagerTest.cs | 6 ++ .../StoringAsyncEndpointExecutorTest.cs | 53 ++++++++++++++++++ 12 files changed, 212 insertions(+), 9 deletions(-) create mode 100644 edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/IEndpointExecutorRetrySignal.cs diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.CloudProxy/DeviceConnectivityManager.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.CloudProxy/DeviceConnectivityManager.cs index 09b18a7a40d..993b0208f44 100644 --- a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.CloudProxy/DeviceConnectivityManager.cs +++ b/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(); } diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Core/IDeviceConnectivityManager.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Core/IDeviceConnectivityManager.cs index 1121b092b97..efd41183d86 100644 --- a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Core/IDeviceConnectivityManager.cs +++ b/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(); diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Core/NullDeviceConnectivityManager.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Core/NullDeviceConnectivityManager.cs index 4764478dc90..108884e2e2f 100644 --- a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Core/NullDeviceConnectivityManager.cs +++ b/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; diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Service/modules/RoutingModule.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Service/modules/RoutingModule.cs index 106fdf063eb..6503eaf0a49 100644 --- a/edge-hub/core/src/Microsoft.Azure.Devices.Edge.Hub.Service/modules/RoutingModule.cs +++ b/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(); var messageStore = await c.Resolve>(); - 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().ConnectivityRecovered += (_, __) => retrySignal.RequestRetry(); + IEndpointExecutorFactory endpointExecutorFactory = new StoringAsyncEndpointExecutorFactory( + endpointExecutorConfig, + new AsyncEndpointExecutorOptions(10, TimeSpan.FromSeconds(10)), + messageStore, + retrySignal); return endpointExecutorFactory; }) .As>() diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/IEndpointExecutorRetrySignal.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/IEndpointExecutorRetrySignal.cs new file mode 100644 index 00000000000..b5f3db026df --- /dev/null +++ b/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); + } +} diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/StoringAsyncEndpointExecutor.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/StoringAsyncEndpointExecutor.cs index 75da4123baa..eb4d715d32f 100644 --- a/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/StoringAsyncEndpointExecutor.cs +++ b/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> 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.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 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 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 snapshot = this.prioritiesToFsms; this.prioritiesToFsms.CompareAndSet(snapshot, ImmutableDictionary.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 diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/StoringAsyncEndpointExecutorFactory.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/StoringAsyncEndpointExecutorFactory.cs index e463bf12cba..f0cfff73957 100644 --- a/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/StoringAsyncEndpointExecutorFactory.cs +++ b/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 CreateAsync(Endpoint endpoint, IList priorities) => this.CreateAsync(endpoint, priorities, new NullCheckpointerFactory(), this.config); @@ -29,7 +35,7 @@ public async Task CreateAsync(Endpoint endpoint, IList 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()); return endpointExecutor; } diff --git a/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/statemachine/EndpointExecutorFsm.cs b/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/statemachine/EndpointExecutorFsm.cs index 0f1b38b6014..4fc0ae5de0f 100644 --- a/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/statemachine/EndpointExecutorFsm.cs +++ b/edge-hub/core/src/Microsoft.Azure.Devices.Routing.Core/endpoints/statemachine/EndpointExecutorFsm.cs @@ -134,6 +134,20 @@ public async Task RunAsync(ICommand command) } } + public async Task RetryNowAsync() + { + using (await this.sync.LockAsync()) + { + if (this.state == State.Failing) + { + this.retryTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); + Routing.UserMetricLogger.LogRetryOperation(1, this.Endpoint.IotHubName, this.Endpoint.Name, this.Endpoint.Type); + Events.RetryNow(this); + await RunInternalAsync(this, Commands.Retry); + } + } + } + public Task CloseAsync() => this.RunAsync(Commands.Close); public void Dispose() => this.Dispose(true); @@ -545,9 +559,7 @@ async void RetryAsync(object obj) { try { - Routing.UserMetricLogger.LogRetryOperation(1, this.Endpoint.IotHubName, this.Endpoint.Name, this.Endpoint.Type); - this.retryTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); - await this.RunAsync(Commands.Retry); + await this.RetryNowAsync(); } catch (Exception ex) { @@ -589,7 +601,8 @@ enum EventIds UpdateEndpoint, UpdateEndpointSuccess, UpdateEndpointFailure, - CheckRetryInnerException + CheckRetryInnerException, + RetryNow } public static void StateEnter(EndpointExecutorFsm fsm) @@ -758,6 +771,11 @@ public static void RetryFailed(EndpointExecutorFsm fsm, Exception exception) Log.LogError((int)EventIds.RetryFailed, exception, "[RetryFailed] Failed to retry. {0}", GetContextString(fsm)); } + public static void RetryNow(EndpointExecutorFsm fsm) + { + Log.LogDebug((int)EventIds.RetryNow, "[RetryNow] Retrying immediately after connectivity recovery. {0}", GetContextString(fsm)); + } + public static void Dead(EndpointExecutorFsm fsm, ICollection messages) { Preconditions.CheckArgument(fsm.Status.LastFailedRevivalTime.HasValue); diff --git a/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/ConnectivityAwareClientTest.cs b/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/ConnectivityAwareClientTest.cs index 87d035a24c0..a052ce12ec4 100644 --- a/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/ConnectivityAwareClientTest.cs +++ b/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/ConnectivityAwareClientTest.cs @@ -353,6 +353,12 @@ class DeviceConnectivityManager : IDeviceConnectivityManager public event EventHandler DeviceDisconnected; + public event EventHandler ConnectivityRecovered + { + add { } + remove { } + } + public Task CallSucceeded() => Task.CompletedTask; public Task CallTimedOut() => Task.CompletedTask; diff --git a/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/DeviceConnectivityManagerTest.cs b/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/DeviceConnectivityManagerTest.cs index 73aa14a77dc..9f15f4217f2 100644 --- a/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/DeviceConnectivityManagerTest.cs +++ b/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test/DeviceConnectivityManagerTest.cs @@ -17,6 +17,29 @@ namespace Microsoft.Azure.Devices.Edge.Hub.CloudProxy.Test [Unit] public class DeviceConnectivityManagerTest { + [Fact] + public async Task ConnectivityRecoveredFiresForTryingToConnectedTransitionTest() + { + var edgeHubIdentity = Mock.Of(i => i.Id == "d1/m1"); + var deviceConnectivityManager = new DeviceConnectivityManager(TimeSpan.FromHours(1), TimeSpan.FromHours(1), edgeHubIdentity); + int connectivityRecovered = 0; + int deviceConnected = 0; + deviceConnectivityManager.ConnectivityRecovered += (_, __) => Interlocked.Increment(ref connectivityRecovered); + deviceConnectivityManager.DeviceConnected += (_, __) => Interlocked.Increment(ref deviceConnected); + + // Initial startup transitions Disconnected -> Connected. + await deviceConnectivityManager.CallSucceeded(); + Assert.Equal(1, connectivityRecovered); + Assert.Equal(1, deviceConnected); + + // A short outage only reaches Trying, not Disconnected. Recovery must still + // wake endpoint retries, without changing the existing DeviceConnected semantics. + await deviceConnectivityManager.CallTimedOut(); + await deviceConnectivityManager.CallSucceeded(); + Assert.Equal(2, connectivityRecovered); + Assert.Equal(1, deviceConnected); + } + [Fact] public async Task NoEventsTest() { diff --git a/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.Core.Test/ConnectionManagerTest.cs b/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.Core.Test/ConnectionManagerTest.cs index b17368a4e7b..64e2de9a823 100644 --- a/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.Core.Test/ConnectionManagerTest.cs +++ b/edge-hub/core/test/Microsoft.Azure.Devices.Edge.Hub.Core.Test/ConnectionManagerTest.cs @@ -1452,6 +1452,12 @@ class DeviceConnectivityManager : IDeviceConnectivityManager public event EventHandler DeviceDisconnected; + public event EventHandler ConnectivityRecovered + { + add { } + remove { } + } + public Task CallSucceeded() => Task.CompletedTask; public Task CallTimedOut() => Task.CompletedTask; diff --git a/edge-hub/core/test/Microsoft.Azure.Devices.Routing.Core.Test/endpoints/StoringAsyncEndpointExecutorTest.cs b/edge-hub/core/test/Microsoft.Azure.Devices.Routing.Core.Test/endpoints/StoringAsyncEndpointExecutorTest.cs index 4c538a73482..1406dbc9af3 100644 --- a/edge-hub/core/test/Microsoft.Azure.Devices.Routing.Core.Test/endpoints/StoringAsyncEndpointExecutorTest.cs +++ b/edge-hub/core/test/Microsoft.Azure.Devices.Routing.Core.Test/endpoints/StoringAsyncEndpointExecutorTest.cs @@ -20,6 +20,59 @@ namespace Microsoft.Azure.Devices.Routing.Core.Test.Endpoints [Integration] public class StoringAsyncEndpointExecutorTest { + [Fact] + public async Task ConnectivityRecoveryInterruptsRetryBackoffTest() + { + const string EndpointId = "endpoint1"; + const uint Priority = 0; + var endpoint = new TestEndpoint(EndpointId) { CanProcess = false }; + var retrySignal = new EndpointExecutorRetrySignal(); + var config = new EndpointExecutorConfig( + TimeSpan.FromSeconds(30), + new FixedInterval(int.MaxValue, TimeSpan.FromMinutes(1)), + TimeSpan.FromSeconds(30)); + var options = new AsyncEndpointExecutorOptions(10, TimeSpan.FromSeconds(10)); + var messageStore = new TestMessageStore(); + var executor = new StoringAsyncEndpointExecutor( + endpoint, + new NullCheckpointerFactory(), + config, + options, + messageStore, + retrySignal); + await executor.UpdatePriorities(new List { Priority }, Option.None()); + + await executor.Invoke(GetNewMessages(1, 0).First(), Priority, 3600); + for (int i = 0; i < 50 && executor.Status.RetryAttempts == 0; i++) + { + await Task.Delay(100); + } + + Assert.Equal(0, endpoint.N); + Assert.True(executor.Status.RetryAttempts > 0); + Assert.Equal(TimeSpan.FromMinutes(1), executor.Status.RetryPeriod); + + endpoint.CanProcess = true; + var sw = System.Diagnostics.Stopwatch.StartNew(); + retrySignal.RequestRetry(); + while (sw.Elapsed < TimeSpan.FromSeconds(5) && endpoint.N == 0) + { + await Task.Delay(50); + } + + Assert.Equal(1, endpoint.N); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(5), $"Reconnect-triggered retry took {sw.Elapsed}."); + + // Repeated connected signals are harmless once the FSM is no longer failing. + retrySignal.RequestRetry(); + await Task.Delay(100); + Assert.Equal(1, endpoint.N); + await executor.CloseAsync(); + + // The executor unsubscribes when closed. + retrySignal.RequestRetry(); + } + [Fact] public async Task InvokeTest() {