diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs index 9a85b14e..0db36c00 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs @@ -440,12 +440,14 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery private readonly ConcurrentDictionary _nativeSubs = new(); private long _nextNativeSubId; + private long _nextRegId; /// // Subscribe via native ADS notifications when TwinCATDriverOptions.UseNativeNotifications is // true, otherwise fall through to the shared PollGroupEngine. Native path registers one - // ITwinCATNotificationHandle per tag against the target's PLC runtime — the PLC pushes - // changes on its own cycle so we skip the poll loop entirely. Unsub path disposes the handles. + // notification per tag against the target's PLC runtime — the PLC pushes changes on its own + // cycle so we skip the poll loop entirely. Each registration stores its REPLAYABLE INTENT on + // the owning device so EnsureConnectedAsync can re-register it after a reconnect (STAB-2). public async Task SubscribeAsync( IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) { @@ -454,8 +456,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery var id = Interlocked.Increment(ref _nextNativeSubId); var handle = new NativeSubscriptionHandle(id); - var registrations = new List(fullReferences.Count); - var now = DateTime.UtcNow; + var registrations = new List(fullReferences.Count); try { @@ -464,18 +465,20 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery if (!_resolver.TryResolve(reference, out var def)) continue; if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) continue; - var client = await EnsureConnectedAsync(device, cancellationToken).ConfigureAwait(false); var parsed = TwinCATSymbolPath.TryParse(def.SymbolPath); var symbolName = parsed?.ToAdsSymbolName() ?? def.SymbolPath; var bitIndex = parsed?.BitIndex; - var reg = await client.AddNotificationAsync( - symbolName, def.DataType, bitIndex, publishingInterval, - _options.NotificationMaxDelayMs, + // The callback closes over the stable (handle, reference) pair, so it survives a + // client swap unchanged — replay re-registers the SAME intent onto the new client. + var reg = new NativeRegistration( + Interlocked.Increment(ref _nextRegId), device, reference, symbolName, + def.DataType, bitIndex, publishingInterval, (_, value) => OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, reference, new DataValueSnapshot( - value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow))), - cancellationToken).ConfigureAwait(false); + value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow)))); + + await RegisterNotificationAsync(device, reg, cancellationToken).ConfigureAwait(false); registrations.Add(reg); } } @@ -496,11 +499,38 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery return handle; } + /// + /// Registers one native notification and records its replayable intent on the device — both + /// under the device so the register can't race a + /// concurrent reconnect-replay (which also runs under that gate). The intent is added to the + /// device registry only AFTER the handle is live, so a replay triggered by this same + /// EnsureConnected can't double-register the not-yet-stored tag. + /// + private async Task RegisterNotificationAsync(DeviceState device, NativeRegistration reg, CancellationToken ct) + { + await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false); + try + { + var client = await EnsureConnectedUnderGateAsync(device, ct, timeoutOverride: null).ConfigureAwait(false); + var newHandle = await client.AddNotificationAsync( + reg.SymbolName, reg.DataType, reg.BitIndex, reg.Interval, + _options.NotificationMaxDelayMs, reg.OnChange, ct).ConfigureAwait(false); + reg.SwapHandle(newHandle); // initial set — previous handle is null + device.NativeRegistrations[reg.Id] = reg; + } + finally + { + device.ConnectGate.Release(); + } + } + /// public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { if (handle is NativeSubscriptionHandle native && _nativeSubs.TryRemove(native.Id, out var sub)) { + // NativeRegistration.Dispose removes itself from its device registry (so a later replay + // won't touch it) and disposes its live handle. foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } } return Task.CompletedTask; } @@ -516,7 +546,64 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery private sealed record NativeSubscription( NativeSubscriptionHandle Handle, - IReadOnlyList Registrations); + IReadOnlyList Registrations); + + /// + /// One live native-notification registration: its replayable intent plus the current live + /// . The handle is swapped (not re-created) when the + /// device reconnects, so unsubscribe / teardown always dispose the latest one. The handle + /// field is exchanged atomically so a reconnect-replay swap and an unsubscribe dispose can't + /// corrupt each other on the common path. + /// + internal sealed class NativeRegistration : IDisposable + { + /// Globally-unique registration id (registry key on the device). + public long Id { get; } + /// The device this notification is registered against. + public DeviceState Device { get; } + /// Driver-side full reference the pushed value is reported under. + public string Reference { get; } + /// ADS symbol name (resolved from the symbol path). + public string SymbolName { get; } + /// Declared data type. + public TwinCATDataType DataType { get; } + /// Bit index for a BOOL-within-word symbol; null otherwise. + public int? BitIndex { get; } + /// Requested notification interval. + public TimeSpan Interval { get; } + /// Value-change callback (closes over the owning subscription handle + reference). + public Action OnChange { get; } + + private ITwinCATNotificationHandle? _handle; + + /// Initializes a new instance of the class. + public NativeRegistration(long id, DeviceState device, string reference, string symbolName, + TwinCATDataType dataType, int? bitIndex, TimeSpan interval, Action onChange) + { + Id = id; + Device = device; + Reference = reference; + SymbolName = symbolName; + DataType = dataType; + BitIndex = bitIndex; + Interval = interval; + OnChange = onChange; + } + + /// Installs a freshly-registered handle and returns the previous one (null on first set). + /// The newly-registered notification handle. + /// The previous handle, or null. + public ITwinCATNotificationHandle? SwapHandle(ITwinCATNotificationHandle newHandle) => + Interlocked.Exchange(ref _handle, newHandle); + + /// Removes this registration from its device and disposes its live handle. + public void Dispose() + { + Device.NativeRegistrations.TryRemove(Id, out _); + var h = Interlocked.Exchange(ref _handle, null); + try { h?.Dispose(); } catch { /* best-effort */ } + } + } // ---- IHostConnectivityProbe ---- @@ -536,12 +623,21 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery var client = await EnsureConnectedAsync(state, ct, _options.Probe.Timeout) .ConfigureAwait(false); success = await client.ProbeAsync(ct).ConfigureAwait(false); + + // A locally-connected client whose wire probe FAILS is wire-dead but port-open: + // force a recycle so the next tick rebuilds + replays native notifications. Without + // this the IsConnected fast-path would keep handing back the dead client forever + // (the STAB-2 compounding bug — native pushes would never recover). + if (!success) + await RecycleClientAsync(state).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; } catch { - // Probe failure — EnsureConnectedAsync's connect-failure path already disposed - // + cleared the client, so next tick will reconnect. + // Probe threw — the connect-failure path clears the client only when the CONNECT + // failed; a throw from ProbeAsync on an already-connected client would leave it in + // place, so force a recycle to guarantee the next tick rebuilds + replays. + await RecycleClientAsync(state).ConfigureAwait(false); } TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped); @@ -608,41 +704,126 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false); try { - // Re-check under the gate: another caller may have connected while we waited. - if (device.Client is { IsConnected: true } c) return c; + return await EnsureConnectedUnderGateAsync(device, ct, timeoutOverride).ConfigureAwait(false); + } + finally + { + device.ConnectGate.Release(); + } + } - // Discard a stale (created-but-disconnected) client before making a fresh one. - if (device.Client is { IsConnected: false } stale) - { - try { stale.Dispose(); } catch { /* best-effort */ } - device.Client = null; - } + /// + /// The connect-or-reconnect core, assuming the caller already holds + /// . When it installs a NEW client (a genuine + /// reconnect, not a fast-path reuse) it replays every stored native-notification intent + /// onto it — otherwise native ADS pushes would silently stop after a drop (STAB-2). Both + /// the initial subscribe () and the probe loop call + /// through here under the same gate, so a replay can never race a register. + /// + private async Task EnsureConnectedUnderGateAsync( + DeviceState device, CancellationToken ct, TimeSpan? timeoutOverride) + { + // Re-check under the gate: another caller may have connected while we waited. + if (device.Client is { IsConnected: true } c) return c; - var client = _clientFactory.Create(); - client.OnSymbolVersionChanged += HandleSymbolVersionChanged; - // timeoutOverride lets the probe loop use TwinCATProbeOptions.Timeout for probe- - // initiated connects rather than the driver-level _options.Timeout. Reads / writes - // pass null and get the driver default. - var effectiveTimeout = timeoutOverride ?? _options.Timeout; + // Discard a stale (created-but-disconnected) client before making a fresh one. + if (device.Client is { IsConnected: false } stale) + { + stale.OnSymbolVersionChanged -= HandleSymbolVersionChanged; + try { stale.Dispose(); } catch { /* best-effort */ } + device.Client = null; + } + + var client = _clientFactory.Create(); + client.OnSymbolVersionChanged += HandleSymbolVersionChanged; + // timeoutOverride lets the probe loop use TwinCATProbeOptions.Timeout for probe- + // initiated connects rather than the driver-level _options.Timeout. Reads / writes + // pass null and get the driver default. + var effectiveTimeout = timeoutOverride ?? _options.Timeout; + try + { + await client.ConnectAsync(device.ParsedAddress, effectiveTimeout, ct) + .ConfigureAwait(false); + _logger.LogInformation( + "TwinCAT driver '{DriverInstanceId}' connected to {HostAddress}", + _driverInstanceId, device.Options.HostAddress); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "TwinCAT driver '{DriverInstanceId}' failed to connect to {HostAddress}", + _driverInstanceId, device.Options.HostAddress); + client.OnSymbolVersionChanged -= HandleSymbolVersionChanged; + client.Dispose(); + throw; + } + device.Client = client; + + // Re-register every native notification onto the fresh client. Native ADS is push, so + // (unlike the poll fallback) it cannot self-heal by re-reading — without this the + // subscription is silently dead after a reconnect. + await ReplayNativeRegistrationsAsync(device, client, ct).ConfigureAwait(false); + return client; + } + + /// + /// Re-registers every stored for a device onto the given + /// freshly-connected client and swaps in the new live handle (disposing the dead one from + /// the previous client). Runs under . A single + /// registration that fails to replay is logged and skipped — one bad symbol must not abort + /// recovery of the rest. + /// + private async Task ReplayNativeRegistrationsAsync(DeviceState device, ITwinCATClient client, CancellationToken ct) + { + var intents = device.NativeRegistrations.Values.ToArray(); + if (intents.Length == 0) return; + + var replayed = 0; + foreach (var reg in intents) + { try { - await client.ConnectAsync(device.ParsedAddress, effectiveTimeout, ct) - .ConfigureAwait(false); - _logger.LogInformation( - "TwinCAT driver '{DriverInstanceId}' connected to {HostAddress}", - _driverInstanceId, device.Options.HostAddress); + var newHandle = await client.AddNotificationAsync( + reg.SymbolName, reg.DataType, reg.BitIndex, reg.Interval, + _options.NotificationMaxDelayMs, reg.OnChange, ct).ConfigureAwait(false); + var old = reg.SwapHandle(newHandle); + try { old?.Dispose(); } catch { /* dead handle from the disposed client */ } + replayed++; } catch (Exception ex) { _logger.LogWarning(ex, - "TwinCAT driver '{DriverInstanceId}' failed to connect to {HostAddress}", - _driverInstanceId, device.Options.HostAddress); - client.OnSymbolVersionChanged -= HandleSymbolVersionChanged; - client.Dispose(); - throw; + "TwinCAT driver '{DriverInstanceId}' failed to replay native notification " + + "'{Symbol}' on device '{HostAddress}' after reconnect", + _driverInstanceId, reg.SymbolName, device.Options.HostAddress); + } + } + + _logger.LogInformation( + "TwinCAT driver '{DriverInstanceId}' re-registered {Replayed}/{Total} native notifications " + + "on device '{HostAddress}' after reconnect", + _driverInstanceId, replayed, intents.Length, device.Options.HostAddress); + } + + /// + /// Disposes + nulls a device's client under so the + /// next rebuilds it (and replays notifications). Called + /// by the probe loop when a wire-level probe fails on a locally-"connected" client — the + /// fast-path keys on the AMS-port state, not wire + /// liveness, so without this forced recycle a wire-dead-but-port-open client would keep + /// being reused and native pushes would never recover. + /// + private async Task RecycleClientAsync(DeviceState device) + { + await device.ConnectGate.WaitAsync(CancellationToken.None).ConfigureAwait(false); + try + { + if (device.Client is { } c) + { + c.OnSymbolVersionChanged -= HandleSymbolVersionChanged; + try { c.Dispose(); } catch { /* best-effort */ } + device.Client = null; } - device.Client = client; - return client; } finally { @@ -722,6 +903,15 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery /// create-or-dispose for this device. public SemaphoreSlim ConnectGate { get; } = new(1, 1); + /// + /// Live native-notification registrations against this device, keyed by registration id. + /// Holds the replayable intent (symbol / type / bit / interval / callback), not just + /// an opaque handle, so that when swaps in + /// a fresh client after a drop it can re-register every notification onto it (STAB-2). + /// Populated by SubscribeAsync, drained by UnsubscribeAsync / teardown. + /// + public ConcurrentDictionary NativeRegistrations { get; } = new(); + /// Gets the lock object for synchronizing host state transitions. public object ProbeLock { get; } = new(); /// Gets or sets the current host connectivity state. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs index 46830b95..40c55cd9 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs @@ -42,6 +42,11 @@ internal class FakeTwinCATClient : ITwinCATClient /// Test hook — fire the symbol-version-changed signal as the real client would. public void FireSymbolVersionChanged() => OnSymbolVersionChanged?.Invoke(this, EventArgs.Empty); + /// Test hook — simulate a wire-level drop that leaves the AMS port "connected" but dead: + /// flips to false without disposing, so the driver's IsConnected + /// fast-path treats the client as stale and rebuilds (exercises the STAB-2 reconnect+replay). + public void SimulateWireDrop() => IsConnected = false; + /// Simulates connecting to the TwinCAT system. /// The AMS address to connect to. /// The connection timeout. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReconnectReplayTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReconnectReplayTests.cs new file mode 100644 index 00000000..d4940620 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReconnectReplayTests.cs @@ -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; + +/// +/// 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 no TwinCAT docker +/// fixture (the 13 integration tests need a real TC3 XAR target), so these fake-client tests +/// are the authoritative automated coverage for the reconnect path. +/// +[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); + } + + /// + /// 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. + /// + [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(); + 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"); + } + + /// Every subscribed tag on the device is replayed, not just the first. + [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); + } + + /// + /// After a reconnect + replay, unsubscribing disposes the CURRENT (replayed) handle and the + /// registration no longer replays on a subsequent reconnect. + /// + [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(); + } + + /// + /// 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). + /// + [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 condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition() && DateTime.UtcNow < deadline) + await Task.Delay(20); + } +}