fix(mtconnect): stamp the /sample pump with its session generation and supersede a stale one (#602 STAB-20)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs

@@ -327,6 +327,20 @@ public sealed class MTConnectDriver
     /// <summary>The shared <c>/sample</c> pump task. Guarded by <see cref="_subscriptionLock"/>.</summary>
     private Task? _pumpTask;
 
+    /// <summary>
+    ///     The <see cref="_sessionGeneration"/> the running pump was started against (STAB-20), guarded by
+    ///     <see cref="_subscriptionLock"/> like the rest of the control block. A pump whose generation is
+    ///     not the current one is bound to a client a lifecycle restart has retired.
+    /// </summary>
+    private long _pumpGeneration;
+
+    /// <summary>
+    ///     Bumped each time <see cref="StartCoreAsync"/> installs a NEW <see cref="AgentSession"/> (not on a
+    ///     <c>with</c>-copy that only advances the cursor or the probe cache). Written under
+    ///     <see cref="_subscriptionLock"/>.
+    /// </summary>
+    private long _sessionGeneration;
+
     /// <summary>
     ///     The Agent answered <c>/sample</c> with something that cannot be streamed at all, so the
     ///     pump gave up. Latched — reconnecting reproduces the identical answer forever, and a
@@ -1808,7 +1822,23 @@ public sealed class MTConnectDriver
         // revived by a resubscribe. The genuinely-unrepeatable case is the latch below, not this.
         if (_pumpTask is { IsCompleted: false })
         {
-            return;
+            if (_pumpGeneration == _sessionGeneration)
+            {
+                return;
+            }
+
+            // STAB-20: a Subscribe can land between StopSampleStreamAsync clearing the registry slots and
+            // the teardown that retires the client — StopSampleStreamAsync is still awaiting the OLD pump
+            // then — and starts a pump on the session about to be disposed. Treating that pump as "the
+            // running stream" would leave the freshly-installed session with no live stream at all
+            // (the stale pump just reads its disposed client as a dropped connection). Retire it: it is
+            // cancelled and can only fail against its disposed client.
+            try { _pumpCts?.Cancel(); }
+            catch (ObjectDisposedException) { /* already going away */ }
+
+            _logger.LogInformation(
+                "MTConnect driver {DriverInstanceId} replaced a /sample pump bound to a retired Agent session.",
+                _driverInstanceId);
         }
 
         if (_streamUnsupported)
@@ -1856,6 +1886,7 @@ public sealed class MTConnectDriver
         var options = Volatile.Read(ref _options);
         var cts = new CancellationTokenSource();
         _pumpCts = cts;
+        _pumpGeneration = _sessionGeneration;
         _pumpTask = Task.Run(
             () => RunSampleStreamAsync(
                 session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, cts.Token),
@@ -2695,6 +2726,11 @@ public sealed class MTConnectDriver
 
             // One publish, so a concurrent lock-free reader sees the new client, the new probe
             // cache, and the pump's opening cursor together or none of them.
+            lock (_subscriptionLock)
+            {
+                _sessionGeneration++;
+            }
+
             Volatile.Write(
                 ref _session,
                 new AgentSession(client, probe, current.InstanceId, current.NextSequence));

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs

@@ -0,0 +1,98 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
+
+/// <summary>
+///     STAB-20 / #602 — a Subscribe that lands between <c>StopSampleStreamAsync</c> clearing the pump slots
+///     and the teardown that retires the client started a pump on the soon-to-be-disposed session. The
+///     lifecycle restart then saw "a pump is running" and never started one against the new session, leaving
+///     the subscription with no live stream. The pump is now stamped with the session generation and a
+///     start request supersedes a pump from an older one.
+/// </summary>
+public sealed class MTConnectPumpGenerationTests
+{
+    private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(10);
+
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+    /// <summary>The pump a mid-restart Subscribe started on the retired client is replaced by one on the new client.</summary>
+    [Fact]
+    public async Task Subscribe_during_a_lifecycle_restart_rebinds_the_pump_to_the_new_client()
+    {
+        var clientA = CannedAgentClient.FromFixtures();
+        var clientB = CannedAgentClient.FromFixtures();
+        var clients = new Queue<CannedAgentClient>([clientA, clientB]);
+
+        var driver = new MTConnectDriver(
+            new MTConnectDriverOptions
+            {
+                AgentUri = "http://fixture-agent:5000",
+                RequestTimeoutMs = 5000,
+                Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
+            },
+            "mt-gen",
+            _ => clients.Dequeue());
+        await driver.InitializeAsync("{}", Ct);
+
+        var blockOnce = 0;
+        var blocking = new ManualResetEventSlim(false);
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        driver.OnDataChange += (_, e) =>
+        {
+            // Once only: the second subscriber's initial-data callback replays the same 1.5 and must not wedge.
+            if (e.Snapshot.Value is not 1.5d || Interlocked.Exchange(ref blockOnce, 1) != 0)
+            {
+                return;
+            }
+
+            entered.TrySetResult();
+            blocking.Wait();
+        };
+
+        try
+        {
+            await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+            _ = clientA.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "1.5"));
+            await entered.Task.WaitAsync(Watchdog, Ct);
+
+            // The first pump is now wedged in caller code. A re-initialize clears the pump slots and then
+            // waits (bounded) on it, which is the window the race lives in.
+            var restart = driver.InitializeAsync("{}", Ct);
+            var spins = 0;
+            while (driver.SampleStreamTask is not null)
+            {
+                await Task.Delay(10, Ct);
+                (++spins).ShouldBeLessThan(500, "restart never cleared the pump slot");
+            }
+
+            // A second subscriber arrives mid-restart: it starts a pump on the (still-live) old client.
+            await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+            await clientA.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
+            restart.IsCompleted.ShouldBeFalse();
+
+            blocking.Set();
+            await restart.WaitAsync(Watchdog, Ct);
+
+            // The restart installed a new client; the standing subscription must be streaming from IT.
+            await clientB.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
+            clientA.IsDisposed.ShouldBeTrue();
+        }
+        finally
+        {
+            blocking.Set();
+            await driver.ShutdownAsync(CancellationToken.None);
+        }
+    }
+}
+
+internal static class MTConnectPumpGenerationChunks
+{
+    public static MTConnectStreamsResult Pos(long first, long next, string value) =>
+        new(
+            1655000000L,
+            next,
+            first,
+            [new MTConnectObservation("dev1_pos", value, new DateTime(2026, 7, 24, 12, 30, 0, DateTimeKind.Utc))]);
+}