fix(twincat): retry failed replay intents on the next successful probe tick (05/STAB-16)

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"
       ]

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);
     }
 
+    /// <summary>
+    ///     05/STAB-16: re-registers every dead (failed-replay) <see cref="NativeRegistration"/> onto
+    ///     the device's currently-connected client. Runs under <see cref="DeviceState.ConnectGate"/>
+    ///     (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.
+    /// </summary>
+    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();
+        }
+    }
+
     /// <summary>
     ///     Disposes + nulls a device's client under <see cref="DeviceState.ConnectGate"/> so the
     ///     next <see cref="EnsureConnectedAsync"/> rebuilds it (and replays notifications). Called

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReconnectReplayTests.cs

@@ -223,6 +223,63 @@ public sealed class TwinCATReconnectReplayTests
         health.LastSuccessfulRead.ShouldBe(lastGoodRead);
     }
 
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<DataChangeEventArgs>();
+        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<bool> condition, TimeSpan timeout)
     {
         var deadline = DateTime.UtcNow + timeout;