fix(twincat): replay native ADS notifications after reconnect (archreview STAB-2 / Critical 4)

Native ADS notifications (the default subscribe mode) were stored as opaque handles with no record of the symbol/type/interval/handler needed to replay. On a client swap (EnsureConnectedAsync building a fresh client after a drop) the notifications were silently orphaned — no Bad status, no error, pushes just stopped until redeploy. Compounding: the IsConnected fast-path keys on AMS-port state, not wire liveness, and a probe failure only transitioned state without recycling the dead client. Fix: - Store REPLAYABLE INTENT: NativeRegistration (symbol/type/bit/interval/onChange + swappable live handle) hung off DeviceState.NativeRegistrations, populated by SubscribeAsync via RegisterNotificationAsync (under ConnectGate). - Split EnsureConnectedAsync into a gate wrapper + EnsureConnectedUnderGateAsync core; when the core installs a NEW client it replays every stored intent onto it and swaps the live handle (disposing the dead one). Register + replay both run under ConnectGate so they can't race. - Probe loop: on a wire-probe failure (false or throw) RecycleClientAsync disposes+nulls the client so the next tick rebuilds + replays — closes the fast-path-keys-on-port-state compounding bug. No TwinCAT docker fixture exists (integration needs a real TC3 XAR), so the fake-client unit tests are the authoritative coverage: - 4 new guards in TwinCATReconnectReplayTests (replay-onto-fresh-client + push reaches OnDataChange + old handle disposed; replay-all-tags; unsubscribe-after-reconnect stops replaying; probe-failure recycles+rebuilds). - Full TwinCAT unit suite 174/174 green; full solution builds 0 errors.
2026-07-08 17:20:10 -04:00
parent 9cad9ed0fc
commit af318fb442
3 changed files with 413 additions and 39 deletions
@@ -42,6 +42,11 @@ internal class FakeTwinCATClient : ITwinCATClient
    /// <summary>Test hook — fire the symbol-version-changed signal as the real client would.</summary>
    public void FireSymbolVersionChanged() => OnSymbolVersionChanged?.Invoke(this, EventArgs.Empty);

+    /// <summary>Test hook — simulate a wire-level drop that leaves the AMS port "connected" but dead:
+    ///     flips <see cref="IsConnected"/> to false without disposing, so the driver's IsConnected
+    ///     fast-path treats the client as stale and rebuilds (exercises the STAB-2 reconnect+replay).</summary>
+    public void SimulateWireDrop() => IsConnected = false;
+
    /// <summary>Simulates connecting to the TwinCAT system.</summary>
    /// <param name="address">The AMS address to connect to.</param>
    /// <param name="timeout">The connection timeout.</param>
@@ -0,0 +1,179 @@
+using System.Collections.Concurrent;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;
+
+/// <summary>
+///     Unit tests for the STAB-2 fix: native ADS notifications are re-registered onto a fresh
+///     client after a reconnect, instead of being silently orphaned. There is <b>no TwinCAT docker
+///     fixture</b> (the 13 integration tests need a real TC3 XAR target), so these fake-client tests
+///     are the <i>authoritative</i> automated coverage for the reconnect path.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class TwinCATReconnectReplayTests
+{
+    private const string Host = "ads://5.23.91.23.1.1:851";
+
+    private static (TwinCATDriver drv, FakeTwinCATClientFactory factory) NewNativeDriver(
+        bool probe = false, TimeSpan? probeInterval = null, params TwinCATTagDefinition[] tags)
+    {
+        var factory = new FakeTwinCATClientFactory();
+        var drv = new TwinCATDriver(new TwinCATDriverOptions
+        {
+            Devices = [new TwinCATDeviceOptions(Host)],
+            Tags = tags,
+            Probe = new TwinCATProbeOptions
+            {
+                Enabled = probe,
+                Interval = probeInterval ?? TimeSpan.FromMilliseconds(30),
+                Timeout = TimeSpan.FromMilliseconds(100),
+            },
+            UseNativeNotifications = true,
+        }, "drv-1", factory);
+        return (drv, factory);
+    }
+
+    /// <summary>
+    ///     After a client drop, the next connection (driven here through a read) re-registers the
+    ///     native notification with the same symbol / type / interval, a push on the fresh client
+    ///     reaches OnDataChange, and the old handle is disposed.
+    /// </summary>
+    [Fact]
+    public async Task Reconnect_replays_native_notifications_onto_the_fresh_client()
+    {
+        var (drv, factory) = NewNativeDriver(
+            tags: new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt));
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+
+        var events = new ConcurrentQueue<DataChangeEventArgs>();
+        drv.OnDataChange += (_, e) => events.Enqueue(e);
+
+        var interval = TimeSpan.FromMilliseconds(250);
+        _ = await drv.SubscribeAsync(["Speed"], interval, TestContext.Current.CancellationToken);
+
+        var client1 = factory.Clients[0];
+        client1.Notifications.Count.ShouldBe(1);
+        var oldHandle = client1.Notifications[0];
+
+        // Simulate a wire-level drop, then trigger a reconnect via a read.
+        client1.SimulateWireDrop();
+        _ = await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken);
+
+        // A brand-new client was built and the notification replayed onto it — same intent.
+        factory.Clients.Count.ShouldBe(2);
+        var client2 = factory.Clients[1];
+        client2.Notifications.Count.ShouldBe(1);
+        var replayed = client2.Notifications[0];
+        replayed.SymbolPath.ShouldBe("MAIN.Speed");
+        replayed.Type.ShouldBe(TwinCATDataType.DInt);
+
+        // The old handle was disposed; a push on the FRESH client reaches OnDataChange.
+        oldHandle.Disposed.ShouldBeTrue();
+        client2.FireNotification("MAIN.Speed", 7777);
+        events.Count.ShouldBe(1);
+        events.Last().Snapshot.Value.ShouldBe(7777);
+        events.Last().FullReference.ShouldBe("Speed");
+    }
+
+    /// <summary>Every subscribed tag on the device is replayed, not just the first.</summary>
+    [Fact]
+    public async Task Reconnect_replays_all_registered_tags()
+    {
+        var (drv, factory) = NewNativeDriver(
+            tags:
+            [
+                new TwinCATTagDefinition("A", Host, "MAIN.A", TwinCATDataType.DInt),
+                new TwinCATTagDefinition("B", Host, "MAIN.B", TwinCATDataType.Real),
+            ]);
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+        _ = await drv.SubscribeAsync(["A", "B"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
+
+        factory.Clients[0].SimulateWireDrop();
+        _ = await drv.ReadAsync(["A"], TestContext.Current.CancellationToken);
+
+        factory.Clients.Count.ShouldBe(2);
+        factory.Clients[1].Notifications.Select(n => n.SymbolPath)
+            .ShouldBe(["MAIN.A", "MAIN.B"], ignoreOrder: true);
+    }
+
+    /// <summary>
+    ///     After a reconnect + replay, unsubscribing disposes the CURRENT (replayed) handle and the
+    ///     registration no longer replays on a subsequent reconnect.
+    /// </summary>
+    [Fact]
+    public async Task Unsubscribe_after_reconnect_disposes_the_current_handle_and_stops_replaying()
+    {
+        var (drv, factory) = NewNativeDriver(
+            tags: new TwinCATTagDefinition("X", Host, "MAIN.X", TwinCATDataType.DInt));
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+        var handle = await drv.SubscribeAsync(["X"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
+
+        // First reconnect → replayed onto client2.
+        factory.Clients[0].SimulateWireDrop();
+        _ = await drv.ReadAsync(["X"], TestContext.Current.CancellationToken);
+        var client2 = factory.Clients[1];
+        client2.Notifications.Count.ShouldBe(1);
+
+        // Unsubscribe disposes the current (client2) handle.
+        await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
+        client2.Notifications.ShouldBeEmpty();
+
+        // A second reconnect must NOT resurrect the unsubscribed notification.
+        client2.SimulateWireDrop();
+        _ = await drv.ReadAsync(["X"], TestContext.Current.CancellationToken);
+        factory.Clients.Count.ShouldBe(3);
+        factory.Clients[2].Notifications.ShouldBeEmpty();
+    }
+
+    /// <summary>
+    ///     A wire-level probe failure on a locally-"connected" client forces a recycle so the next
+    ///     tick rebuilds the client — the compounding half of STAB-2 (the IsConnected fast-path
+    ///     keys on port state, not wire liveness).
+    /// </summary>
+    [Fact]
+    public async Task Probe_failure_recycles_the_client_and_rebuilds()
+    {
+        // First client's wire probe fails; the recycle nulls it and the next tick builds a second
+        // client whose probe succeeds.
+        var buildCount = 0;
+        var factory = new FakeTwinCATClientFactory
+        {
+            Customise = () =>
+            {
+                buildCount++;
+                // Client #1 reports a failed wire probe; #2+ succeed.
+                return new FakeTwinCATClient { ProbeResult = buildCount > 1 };
+            },
+        };
+        var drv = new TwinCATDriver(new TwinCATDriverOptions
+        {
+            Devices = [new TwinCATDeviceOptions(Host)],
+            Tags = [new TwinCATTagDefinition("X", Host, "MAIN.X", TwinCATDataType.DInt)],
+            Probe = new TwinCATProbeOptions
+            {
+                Enabled = true,
+                Interval = TimeSpan.FromMilliseconds(30),
+                Timeout = TimeSpan.FromMilliseconds(100),
+            },
+            UseNativeNotifications = true,
+        }, "drv-probe", factory);
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+
+        // The probe loop should recycle the wire-dead client and rebuild a healthy one.
+        await WaitForAsync(() => factory.Clients.Count >= 2, TimeSpan.FromSeconds(3));
+        factory.Clients.Count.ShouldBeGreaterThanOrEqualTo(2);
+        factory.Clients[0].DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
+
+        await drv.ShutdownAsync(TestContext.Current.CancellationToken);
+    }
+
+    private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (!condition() && DateTime.UtcNow < deadline)
+            await Task.Delay(20);
+    }
+}