From e76df0741a89bb8970dc001d8392e09d5822e71d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 07:24:43 -0400 Subject: [PATCH] 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 --- .../MTConnectDriver.cs | 38 ++++++- .../MTConnectPumpGenerationTests.cs | 98 +++++++++++++++++++ 2 files changed, 135 insertions(+), 1 deletion(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs 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))]); +}