From de297596431935a3c22b740c03a47f14c3e8e7a4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 12:37:20 -0400 Subject: [PATCH] fix(twincat): retry failed replay intents on the next successful probe tick (05/STAB-16) --- ...2-09-driver-fleet-batch-plan.md.tasks.json | 2 +- .../TwinCATDriver.cs | 63 +++++++++++++++++++ .../TwinCATReconnectReplayTests.cs | 57 +++++++++++++++++ 3 files changed, 121 insertions(+), 1 deletion(-) diff --git a/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json b/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json index 93205f6b..587d853a 100644 --- a/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json +++ b/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json @@ -200,7 +200,7 @@ { "id": "B6.3", "subject": "B6: RetryDeadRegistrationsAsync on successful probe tick (under ConnectGate), tests FAIL->PASS \u2014 STAB-16", - "status": "pending", + "status": "completed", "blockedBy": [ "B6.2" ] 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 e0a7ffe0..bfe45438 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs @@ -690,6 +690,10 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery // (the compounding bug — native pushes would never recover). if (!success) await RecycleClientAsync(state).ConfigureAwait(false); + else + // 05/STAB-16: on a healthy tick, re-register any notifications whose replay + // failed (a transient AddNotification blip) so they recover without a client swap. + await RetryDeadRegistrationsAsync(state, ct).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; } catch @@ -901,6 +905,65 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery _driverInstanceId, replayed, intents.Length, device.Options.HostAddress); } + /// + /// 05/STAB-16: re-registers every dead (failed-replay) onto + /// the device's currently-connected client. Runs under + /// (shared with the replay so it can't race a register/unsubscribe). Called from the probe + /// loop after a successful probe — a bounded retry cadence (the probe interval) with no extra + /// timers. A registration that fails again just stays dead for the next tick. Recovery of the + /// value itself arrives via ADS's initial-notification push on register, exactly as the + /// initial subscribe relies on. + /// + private async Task RetryDeadRegistrationsAsync(DeviceState device, CancellationToken ct) + { + if (device.NativeRegistrations.IsEmpty) return; + + await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false); + try + { + if (device.Client is not { IsConnected: true } client) return; + + var recovered = 0; + foreach (var reg in device.NativeRegistrations.Values) + { + if (reg.HasLiveHandle) continue; + try + { + var newHandle = await client.AddNotificationAsync( + reg.SymbolName, reg.DataType, reg.BitIndex, reg.Interval, + _options.NotificationMaxDelayMs, reg.OnChange, ct).ConfigureAwait(false); + reg.SwapHandle(newHandle); + // STAB-17 ownership re-check — see ReplayNativeRegistrationsAsync. + if (!device.NativeRegistrations.ContainsKey(reg.Id)) + { + var orphan = reg.MarkHandleDead(); + try { orphan?.Dispose(); } catch { /* best-effort */ } + continue; + } + recovered++; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; } + catch (Exception ex) + { + _logger.LogDebug(ex, + "TwinCAT driver '{DriverInstanceId}' probe-tick retry of dead notification " + + "'{Symbol}' on device '{HostAddress}' failed; will retry next tick", + _driverInstanceId, reg.SymbolName, device.Options.HostAddress); + } + } + + if (recovered > 0) + _logger.LogInformation( + "TwinCAT driver '{DriverInstanceId}' recovered {Recovered} dead native notification(s) " + + "on device '{HostAddress}' via probe-tick retry", + _driverInstanceId, recovered, device.Options.HostAddress); + } + finally + { + device.ConnectGate.Release(); + } + } + /// /// Disposes + nulls a device's client under so the /// next rebuilds it (and replays notifications). Called 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 index 44724b8c..30b6f4f7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReconnectReplayTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReconnectReplayTests.cs @@ -223,6 +223,63 @@ public sealed class TwinCATReconnectReplayTests health.LastSuccessfulRead.ShouldBe(lastGoodRead); } + /// + /// STAB-16: a registration whose replay failed (dead handle) is re-registered on the next + /// successful probe tick — a subsequent push then reaches OnDataChange with Good, and no + /// duplicate registration is created. + /// + [Fact] + public async Task ReplayFailure_RetriedOnNextSuccessfulProbeTick() + { + var build = 0; + var factory = new FakeTwinCATClientFactory + { + // Only the 2nd client (the reconnect target) fails its first replay; flipped off below. + Customise = () => new FakeTwinCATClient { ThrowOnAddNotification = ++build == 2 }, + }; + var drv = new TwinCATDriver(new TwinCATDriverOptions + { + Devices = [new TwinCATDeviceOptions(Host)], + Tags = [new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt)], + Probe = new TwinCATProbeOptions + { + Enabled = true, + Interval = TimeSpan.FromMilliseconds(40), + Timeout = TimeSpan.FromMilliseconds(100), + }, + UseNativeNotifications = true, + }, "drv-1", factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + var events = new ConcurrentQueue(); + drv.OnDataChange += (_, e) => events.Enqueue(e); + + _ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(250), TestContext.Current.CancellationToken); + await WaitForAsync(() => factory.Clients.Count >= 1, TimeSpan.FromSeconds(2)); + + // Drop the wire; the reconnect target (client2) fails its replay, leaving a dead registration. + factory.Clients[0].SimulateWireDrop(); + _ = await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken); + await WaitForAsync(() => factory.Clients.Count >= 2, TimeSpan.FromSeconds(2)); + var client2 = factory.Clients[1]; + client2.Notifications.ShouldBeEmpty(); // replay failed — no live handle + + // Device now accepts registrations again — the probe-tick retry must re-register the intent. + client2.ThrowOnAddNotification = false; + await WaitForAsync(() => client2.Notifications.Count == 1, TimeSpan.FromSeconds(3)); + client2.Notifications.Count.ShouldBe(1); // re-registered exactly once (no duplicate) + + // A fresh push on the recovered registration reaches OnDataChange with Good. + client2.FireNotification("MAIN.Speed", 4242); + await WaitForAsync(() => events.Any(e => e.FullReference == "Speed" + && e.Snapshot.StatusCode == TwinCATStatusMapper.Good), TimeSpan.FromSeconds(2)); + events.Any(e => e.FullReference == "Speed" + && e.Snapshot.StatusCode == TwinCATStatusMapper.Good + && Equals(e.Snapshot.Value, 4242)).ShouldBeTrue(); + + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + } + private static async Task WaitForAsync(Func condition, TimeSpan timeout) { var deadline = DateTime.UtcNow + timeout;