diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs index 42c32655..e22e4769 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs @@ -105,14 +105,16 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection // Background event loop: route each value change to the matching subscription callback. _eventLoopCts = new CancellationTokenSource(); - _ = Task.Run(() => RunEventLoopAsync(_eventLoopCts.Token)); + var loopToken = _eventLoopCts.Token; // capture NOW: the field may be replaced by a + var loopClient = _client; // subsequent reconnect before Task.Run executes (N1) + _ = Task.Run(() => RunEventLoopAsync(loopClient, loopToken)); } - private async Task RunEventLoopAsync(CancellationToken ct) + private async Task RunEventLoopAsync(IMxGatewayClient client, CancellationToken ct) { try { - await _client!.RunEventLoopAsync(update => + await client.RunEventLoopAsync(update => { if (_tagToSub.TryGetValue(update.TagPath, out var subId) && _subs.TryGetValue(subId, out var s)) s.Callback(update.TagPath, new TagValue(update.Value, update.Quality, update.Timestamp)); @@ -122,6 +124,18 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection { // Normal shutdown (DisconnectAsync / DisposeAsync cancelled the loop). } + catch (Exception ex) when (ct.IsCancellationRequested) + { + // Teardown fault of a superseded loop (N1): Cancel() is not synchronous with + // the loop's exit — a cancelled/disposed gRPC stream commonly surfaces as + // RpcException(StatusCode.Cancelled) (Grpc.Net default without + // ThrowOperationCanceledOnCancellation) or ObjectDisposedException from the + // concurrent DisposeAsync, milliseconds AFTER ConnectAsync has already reset + // _disconnectFired. Any fault on a cancelled token is normal shutdown; + // raising Disconnected here would flap the NEW healthy session. + _logger.LogDebug(ex, + "MxGateway event loop faulted after cancellation — superseded loop teardown; not signalling disconnect"); + } catch (Exception ex) { _logger.LogWarning(ex, "MxGateway event stream faulted; signalling disconnect"); diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionReconnectTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionReconnectTests.cs index cb032d61..b50cbfd3 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionReconnectTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionReconnectTests.cs @@ -1,3 +1,4 @@ +using Grpc.Core; using Microsoft.Extensions.Logging.Abstractions; using NSubstitute; using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; @@ -47,6 +48,73 @@ public class MxGatewayDataConnectionReconnectTests Assert.False(loopTokens[1].IsCancellationRequested); // new loop live } + [Fact] + public async Task StaleEventLoopRpcFault_AfterReconnect_DoesNotSignalDisconnected() + { + var loops = new List(); + var loopTokens = new List(); + var adapter = CreateAdapterWithControllableLoops(loops, loopTokens, out var disconnectsRef); + + var details = new Dictionary { ["Endpoint"] = "http://gw:5000", ["ApiKey"] = "k" }; + + await adapter.ConnectAsync(details); + await WaitUntilAsync(() => loopTokens.Count == 1); + await adapter.ConnectAsync(details); // reconnect: cancels loop #0, resets _disconnectFired + await WaitUntilAsync(() => loopTokens.Count == 2); + + // The OLD loop observes its cancellation as a gRPC fault, not an OCE — + // the Grpc.Net default without ThrowOperationCanceledOnCancellation. + loops[0].SetException(new RpcException(new Status(StatusCode.Cancelled, "call cancelled"))); + await Task.Delay(200); + Assert.Equal(0, Volatile.Read(ref disconnectsRef.Value)); // stale fault absorbed — the fresh connection must not flap + + // The CURRENT loop's genuine fault must still signal, exactly once. + loops[1].SetException(new RpcException(new Status(StatusCode.Unavailable, "gateway gone"))); + await WaitUntilAsync(() => Volatile.Read(ref disconnectsRef.Value) == 1); + } + + [Fact] + public async Task StaleEventLoopObjectDisposedFault_AfterReconnect_DoesNotSignalDisconnected() + { + var loops = new List(); + var loopTokens = new List(); + var adapter = CreateAdapterWithControllableLoops(loops, loopTokens, out var disconnectsRef); + + var details = new Dictionary { ["Endpoint"] = "http://gw:5000", ["ApiKey"] = "k" }; + + await adapter.ConnectAsync(details); + await WaitUntilAsync(() => loopTokens.Count == 1); + await adapter.ConnectAsync(details); // reconnect + await WaitUntilAsync(() => loopTokens.Count == 2); + + // The old loop faults with the concurrent-DisposeAsync shape. + loops[0].SetException(new ObjectDisposedException("MxGatewayClient")); + await Task.Delay(200); + Assert.Equal(0, Volatile.Read(ref disconnectsRef.Value)); + } + + private sealed class IntRef { public int Value; } + + private static MxGatewayDataConnection CreateAdapterWithControllableLoops( + List loops, List loopTokens, out IntRef disconnectsRef) + { + var factory = Substitute.For(); + factory.Create().Returns(_ => + { + var c = Substitute.For(); + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + loops.Add(tcs); + c.RunEventLoopAsync(Arg.Any>(), Arg.Do(t => loopTokens.Add(t))) + .Returns(tcs.Task); + return c; + }); + var adapter = new MxGatewayDataConnection(factory, NullLogger.Instance); + var refHolder = new IntRef(); + adapter.Disconnected += () => Interlocked.Increment(ref refHolder.Value); + disconnectsRef = refHolder; + return adapter; + } + private static async Task WaitUntilAsync(Func condition) { for (var i = 0; i < 100 && !condition(); i++)