diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
index 5001c1aa..3da7bf61 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
@@ -327,6 +327,20 @@ public sealed class MTConnectDriver
/// The shared /sample pump task. Guarded by .
private Task? _pumpTask;
+ ///
+ /// The the running pump was started against (STAB-20), guarded by
+ /// 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.
+ ///
+ private long _pumpGeneration;
+
+ ///
+ /// Bumped each time installs a NEW (not on a
+ /// with-copy that only advances the cursor or the probe cache). Written under
+ /// .
+ ///
+ private long _sessionGeneration;
+
///
/// The Agent answered /sample 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));
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
new file mode 100644
index 00000000..738f7a5f
--- /dev/null
+++ b/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;
+
+///
+/// STAB-20 / #602 — a Subscribe that lands between StopSampleStreamAsync 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.
+///
+public sealed class MTConnectPumpGenerationTests
+{
+ private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(10);
+
+ private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+ /// The pump a mid-restart Subscribe started on the retired client is replaced by one on the new client.
+ [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([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))]);
+}