fix(dcl): stale MxGateway event-loop fault on a cancelled token no longer flaps the fresh connection (plan R2-03 T1)

2026-07-13 09:45:20 -04:00
parent 1429ddaa0e
commit f25812fc7c
2 changed files with 85 additions and 3 deletions
@@ -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<TaskCompletionSource>();
+        var loopTokens = new List<CancellationToken>();
+        var adapter = CreateAdapterWithControllableLoops(loops, loopTokens, out var disconnectsRef);
+
+        var details = new Dictionary<string, string> { ["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<TaskCompletionSource>();
+        var loopTokens = new List<CancellationToken>();
+        var adapter = CreateAdapterWithControllableLoops(loops, loopTokens, out var disconnectsRef);
+
+        var details = new Dictionary<string, string> { ["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<TaskCompletionSource> loops, List<CancellationToken> loopTokens, out IntRef disconnectsRef)
+    {
+        var factory = Substitute.For<IMxGatewayClientFactory>();
+        factory.Create().Returns(_ =>
+        {
+            var c = Substitute.For<IMxGatewayClient>();
+            var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+            loops.Add(tcs);
+            c.RunEventLoopAsync(Arg.Any<Action<MxValueUpdate>>(), Arg.Do<CancellationToken>(t => loopTokens.Add(t)))
+                .Returns(tcs.Task);
+            return c;
+        });
+        var adapter = new MxGatewayDataConnection(factory, NullLogger<MxGatewayDataConnection>.Instance);
+        var refHolder = new IntRef();
+        adapter.Disconnected += () => Interlocked.Increment(ref refHolder.Value);
+        disconnectsRef = refHolder;
+        return adapter;
+    }
+
    private static async Task WaitUntilAsync(Func<bool> condition)
    {
        for (var i = 0; i < 100 && !condition(); i++)