fix(dcl): cancel old event loop + dispose old MxGateway client on reconnect (S1)

2026-07-08 23:19:11 -04:00
parent 11eb5157e6
commit a8a01f026b
2 changed files with 77 additions and 0 deletions
@@ -67,6 +67,27 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
    public async Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
    {
        var cfg = MxGatewayEndpointConfigSerializer.FromFlatDict(connectionDetails);
+
+        // Reconnect reuses this adapter: cancel the previous event loop and dispose
+        // the previous client so a flapping gateway cannot accumulate orphaned
+        // streams/sockets, and the old loop's fault path cannot signal Disconnected
+        // against the new session after the guard resets below. Keep this BEFORE the
+        // _disconnectFired reset so a fault raised by the dying loop during teardown
+        // is absorbed by the old guard cycle. Dispose is best-effort fire-and-forget.
+        if (_eventLoopCts is { } previousLoopCts)
+        {
+            previousLoopCts.Cancel();
+            previousLoopCts.Dispose();
+            _eventLoopCts = null;
+        }
+        if (_client is { } previousClient)
+        {
+            _ = previousClient.DisposeAsync().AsTask().ContinueWith(
+                t => _logger.LogWarning(t.Exception?.GetBaseException(),
+                    "Disposing previous MxGateway client failed"),
+                TaskContinuationOptions.OnlyOnFaulted);
+        }
+
        Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA
        _client = _clientFactory.Create();

@@ -0,0 +1,56 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
+
+/// <summary>
+/// DataConnectionLayer-S1 (MxGateway half): the adapter instance is reused across
+/// reconnects. Before the fix, <c>ConnectAsync</c> overwrote <c>_client</c> and
+/// <c>_eventLoopCts</c> without cancelling the previous event loop or disposing the
+/// previous client — every gateway flap accumulated an orphaned stream/socket, and
+/// the dying old loop's fault path could raise <c>Disconnected</c> against the new
+/// session after the guard was reset.
+/// </summary>
+[Collection("DataConnectionManagerActor")]
+public class MxGatewayDataConnectionReconnectTests
+{
+    [Fact]
+    public async Task Reconnect_DisposesPreviousClient_AndCancelsPreviousEventLoop()
+    {
+        var clients = new List<IMxGatewayClient>();
+        var loopTokens = new List<CancellationToken>();
+        var factory = Substitute.For<IMxGatewayClientFactory>();
+        factory.Create().Returns(_ =>
+        {
+            var c = Substitute.For<IMxGatewayClient>();
+            c.RunEventLoopAsync(Arg.Any<Action<MxValueUpdate>>(), Arg.Do<CancellationToken>(t => loopTokens.Add(t)))
+                .Returns(Task.Delay(Timeout.Infinite));
+            clients.Add(c);
+            return c;
+        });
+        var adapter = new MxGatewayDataConnection(factory, NullLogger<MxGatewayDataConnection>.Instance);
+        var details = new Dictionary<string, string> { ["Endpoint"] = "http://gw:5000", ["ApiKey"] = "k" };
+
+        // The event loop is pumped fire-and-forget (ConnectAsync → Task.Run →
+        // RunEventLoopAsync), so the token is recorded on a thread-pool continuation.
+        // Barrier on each recorded token before proceeding.
+        await adapter.ConnectAsync(details);
+        await WaitUntilAsync(() => loopTokens.Count == 1);
+
+        await adapter.ConnectAsync(details); // reconnect
+        await WaitUntilAsync(() => loopTokens.Count == 2);
+
+        Assert.Equal(2, clients.Count);
+        await clients[0].Received(1).DisposeAsync();
+        Assert.True(loopTokens[0].IsCancellationRequested);   // old loop cancelled
+        Assert.False(loopTokens[1].IsCancellationRequested);  // new loop live
+    }
+
+    private static async Task WaitUntilAsync(Func<bool> condition)
+    {
+        for (var i = 0; i < 100 && !condition(); i++)
+            await Task.Delay(50);
+        Assert.True(condition(), "condition not met within timeout");
+    }
+}