From 77bfadbdda5675a6cfe082d826da7ca2963b8540 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:36:24 -0400 Subject: [PATCH] fix(mtconnect): generation guard on the pump's cursor publish, chunks and re-baseline (R06:T-1) A pump the lifecycle retired or abandoned that finished after the restart installed the new session published its stale cursor onto it (PublishAgentCursor only checked for a null or identical session), so the next pump read its first chunk as an Agent restart: index cleared, a Bad flicker on every subscribed tag, a needless re-baseline. The pump now carries the session generation it was started against; the cursor publish is checked under _subscriptionLock (where the session install and generation bump are atomic) and a superseded pump also stops folding late chunks / re-baselines into the new session's index. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/MTConnect.md | 7 ++ .../MTConnectDriver.cs | 77 +++++++++--- .../MTConnectStaleCursorTests.cs | 110 ++++++++++++++++++ 3 files changed, 178 insertions(+), 16 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStaleCursorTests.cs diff --git a/docs/drivers/MTConnect.md b/docs/drivers/MTConnect.md index aa6d614e..fe22a648 100644 --- a/docs/drivers/MTConnect.md +++ b/docs/drivers/MTConnect.md @@ -324,6 +324,13 @@ the two need disambiguating, not just OR-ing together). On a changed `/hosts` page — see [Known limitations](#known-limitations) for what that does and does not do. +**A pump carries the session generation it was started against (review 2026-10-21 T-1).** A lifecycle restart +retires the running pump (or abandons it after its bounded teardown wait); if that pump finishes after the new +session is installed, its cursor publish, a late chunk and a late re-baseline are all dropped — checked under the +same lock the new session is installed and the generation bumped under. It used to land the OLD Agent's +`instanceId` + sequence on the new session, so the next pump read its first chunk as an Agent restart (index +cleared, one Bad flicker on every subscribed tag, a needless re-baseline). + ## Write-back — observation injection, off by default > ⚠️ **A write sets what the Agent *reports*, not what the machine does.** The value 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 b6860b53..cbbdb266 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs @@ -1856,9 +1856,10 @@ public sealed partial class MTConnectDriver var cts = new CancellationTokenSource(); _pumpCts = cts; _pumpGeneration = _sessionGeneration; + var generation = _sessionGeneration; _pumpTask = Task.Run( () => RunSampleStreamAsync( - session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, cts.Token), + session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, generation, cts.Token), CancellationToken.None); } @@ -1894,6 +1895,7 @@ public sealed partial class MTConnectDriver long from, long instanceId, bool republishOnStart, + long generation, CancellationToken ct) { var cursor = from; @@ -1929,7 +1931,7 @@ public sealed partial class MTConnectDriver } } - var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, ct).ConfigureAwait(false); + var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, generation, ct).ConfigureAwait(false); cursor = outcome.Cursor; instanceId = outcome.InstanceId; @@ -2015,8 +2017,9 @@ public sealed partial class MTConnectDriver // would therefore reopen at a sequence the Agent has long since evicted: it self-heals // via OUT_OF_RANGE, but only after a wasted round trip, and a still-buffered stale // cursor replays historical observations to subscribers first. One publish per pump - // lifetime, dropped if a lifecycle change has already installed a newer session. - PublishAgentCursor(instanceId, cursor); + // lifetime, dropped if a lifecycle change has already installed a newer session (R06:T-1: the + // generation this pump was started against is what says so — the session alone cannot). + PublishAgentCursor(generation, instanceId, cursor); } } @@ -2030,6 +2033,7 @@ public sealed partial class MTConnectDriver MTConnectDriverOptions options, long cursor, long instanceId, + long generation, CancellationToken ct) { // The RUNNING cursor the gap check is made against: the previous chunk's nextSequence, equal @@ -2052,6 +2056,14 @@ public sealed partial class MTConnectDriver continue; } + // R06:T-1 — a pump whose session a lifecycle restart has replaced (it was retired, or its + // teardown wait was abandoned) must not fold a late chunk of the OLD Agent connection into + // the NEW session's index — nor read it as an Agent restart and clear that index. + if (!IsCurrentSessionGeneration(generation)) + { + return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null); + } + // CHECKED BEFORE THE GAP: an Agent restart changes instanceId *and* resets // sequences, so a restart usually trips IsSequenceGap too — and the two need // different handling. A gap keeps the held values (they are still this device's); @@ -2076,7 +2088,7 @@ public sealed partial class MTConnectDriver // DiscoverAsync. AnnounceAgentRestart(chunk.InstanceId, options); - return await RebaselineAsync(client, options, expected, instanceId, delivered, ct) + return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct) .ConfigureAwait(false); } @@ -2086,7 +2098,7 @@ public sealed partial class MTConnectDriver "MTConnect driver {DriverInstanceId} fell out of {AgentUri}'s buffer (expected sequence {Expected}, oldest retained {FirstSequence}); re-baselining from /current.", _driverInstanceId, options.AgentUri, expected, chunk.FirstSequence); - return await RebaselineAsync(client, options, expected, instanceId, delivered, ct) + return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct) .ConfigureAwait(false); } @@ -2136,7 +2148,7 @@ public sealed partial class MTConnectDriver "MTConnect driver {DriverInstanceId} could not resume {AgentUri}'s /sample stream at sequence {Cursor} (the Agent rejected it as out of range); re-baselining from /current.", _driverInstanceId, options.AgentUri, cursor); - return await RebaselineAsync(client, options, expected, instanceId, delivered, ct) + return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct) .ConfigureAwait(false); } catch (Exception ex) @@ -2166,6 +2178,7 @@ public sealed partial class MTConnectDriver long fallbackCursor, long fallbackInstanceId, bool deliveredAny, + long generation, CancellationToken ct) { try @@ -2173,7 +2186,13 @@ public sealed partial class MTConnectDriver var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, ct) .ConfigureAwait(false); - PublishAgentCursor(current.InstanceId, current.NextSequence); + if (!IsCurrentSessionGeneration(generation)) + { + return new StreamOutcome( + StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, deliveredAny, null); + } + + PublishAgentCursor(generation, current.InstanceId, current.NextSequence); ApplyAndFanOut(current, baseline: true); _streamDegraded = false; @@ -2488,25 +2507,51 @@ public sealed partial class MTConnectDriver /// /// The Agent instanceId last observed. /// The sequence a /sample stream should resume from. - private void PublishAgentCursor(long instanceId, long nextSequence) + /// The session generation the publishing pump was started against. + private void PublishAgentCursor(long generation, long instanceId, long nextSequence) { - while (true) + // R06:T-1 — under _subscriptionLock, the lock StartCoreAsync installs a new session AND bumps the + // generation under: inside it the two are always observed together, so a pump bound to a retired + // session can never land its cursor on the session that replaced it (a lock-free generation check + // could read the new session beside the not-yet-bumped generation). Nothing is awaited here. + lock (_subscriptionLock) { - var session = Volatile.Read(ref _session); - if (session is null || - (session.InstanceId == instanceId && session.NextSequence == nextSequence)) + if (_sessionGeneration != generation) { return; } - var updated = session with { InstanceId = instanceId, NextSequence = nextSequence }; - if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session)) + while (true) { - return; + var session = Volatile.Read(ref _session); + if (session is null || + (session.InstanceId == instanceId && session.NextSequence == nextSequence)) + { + return; + } + + // Still a compare-and-swap: InvalidateProbeModel / GetOrFetchProbeModelAsync swap the session + // lock-free, and a plain write here could undo a probe-model drop. + var updated = session with { InstanceId = instanceId, NextSequence = nextSequence }; + if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session)) + { + return; + } } } } + /// Whether is still the installed session's generation. + /// A pump's session generation. + /// once a lifecycle restart has installed a newer session. + private bool IsCurrentSessionGeneration(long generation) + { + lock (_subscriptionLock) + { + return _sessionGeneration == generation; + } + } + /// /// The /probe device model: the cached copy when warm, otherwise a fresh /// deadline-bounded /probe that re-populates the cache. This is the accessor every diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStaleCursorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStaleCursorTests.cs new file mode 100644 index 00000000..0f8ea7b8 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStaleCursorTests.cs @@ -0,0 +1,110 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests; + +/// +/// Review 2026-10-21 R06:T-1 — a pump the lifecycle abandoned (its teardown wait timed out) that finished after +/// the restart installed the NEW session published its stale cursor onto it: the old Agent's instanceId and +/// sequence replaced the new session's, so the next pump to start treated its first chunk as an Agent restart +/// (index cleared, a Bad flicker on every subscribed tag, a needless re-baseline). The pump now carries the +/// session generation it was started against and its cursor publish is dropped once that generation is gone. +/// +public sealed class MTConnectStaleCursorTests +{ + private const long InstanceB = 1655000099L; + private const long NextSequenceB = 500L; + private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + /// The abandoned pump's late cursor publish leaves the new session's cursor alone. + [Fact] + public async Task An_abandoned_pump_does_not_publish_its_cursor_onto_the_new_session() + { + var clientA = CannedAgentClient.FromFixtures(); + var clientB = CannedAgentClient.FromFixtures(); + clientB.Current = new MTConnectStreamsResult(InstanceB, NextSequenceB, NextSequenceB - 10, clientB.Current.Observations); + 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-t1", + _ => clients.Dequeue()); + await driver.InitializeAsync("{}", Ct); + + var wedgeOnce = 0; + var wedge = new ManualResetEventSlim(false); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + driver.OnDataChange += (_, e) => + { + if (e.Snapshot.Value is not 1.5d || Interlocked.Exchange(ref wedgeOnce, 1) != 0) return; + entered.TrySetResult(); + wedge.Wait(); + }; + + try + { + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + var pumpA = driver.SampleStreamTask.ShouldNotBeNull(); + _ = clientA.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "1.5")); + await entered.Task.WaitAsync(Watchdog, Ct); + + // The restart gives up on the wedged pump (bounded teardown wait), retires clientA and installs B. + await driver.InitializeAsync("{}", Ct).WaitAsync(Watchdog, Ct); + driver.AgentInstanceId.ShouldBe(InstanceB, "positive control: the new session is installed"); + pumpA.IsCompleted.ShouldBeFalse("pump A must still be alive — it is the late writer under test"); + + // Now let the abandoned pump finish: its finally publishes (instanceA, 113). + wedge.Set(); + await pumpA.WaitAsync(Watchdog, Ct); + + driver.AgentInstanceId.ShouldBe(InstanceB, "a superseded pump must not write the old Agent's instanceId"); + driver.AgentNextSequence.ShouldBe(NextSequenceB, "nor its cursor"); + } + finally + { + wedge.Set(); + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// + /// Positive control: the CURRENT pump's own cursor publish still lands — a last unsubscribe stops it, and the + /// session's cursor moves to where that pump got to. + /// + [Fact] + public async Task The_current_pump_still_publishes_its_cursor_when_it_stops() + { + var client = CannedAgentClient.FromFixtures(); + var driver = new MTConnectDriver( + new MTConnectDriverOptions + { + AgentUri = "http://fixture-agent:5000", + RequestTimeoutMs = 5000, + Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)], + }, + "mt-t1-ok", + _ => client); + await driver.InitializeAsync("{}", Ct); + try + { + var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + await client.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "1.5")).WaitAsync(Watchdog, Ct); + + await driver.UnsubscribeAsync(handle, Ct); + + driver.AgentNextSequence.ShouldBe(113); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } +}