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 e7908a08..ed5a4c2e 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
@@ -2207,8 +2207,10 @@ public sealed partial class MTConnectDriver
_streamUnsupported = true;
}
- _streamDegraded = true;
- Degrade(outcome.Error!);
+ if (!TryDegradeStream(generation, outcome.Error!.Message))
+ {
+ return;
+ }
_logger.LogError(
outcome.Error,
@@ -2237,8 +2239,10 @@ public sealed partial class MTConnectDriver
break;
default:
- _streamDegraded = true;
- Degrade(outcome.Error!);
+ if (!TryDegradeStream(generation, outcome.Error!.Message))
+ {
+ return;
+ }
// A stream that DELIVERED before it dropped proves the endpoint works, so
// its failure starts a fresh backoff ladder. Without this the counter is
@@ -2264,11 +2268,7 @@ public sealed partial class MTConnectDriver
"MTConnect driver {DriverInstanceId} stopped its /sample pump on an unexpected fault; subscriptions will receive no further updates until the driver is re-initialized.",
_driverInstanceId);
- if (IsCurrentSessionGeneration(generation))
- {
- _streamDegraded = true;
- Degrade(ex);
- }
+ TryDegradeStream(generation, ex.Message);
}
finally
{
@@ -2347,9 +2347,16 @@ public sealed partial class MTConnectDriver
// Dropped BEFORE the re-baseline, and the subscribers are told: if the /current
// then fails, holding stale values from a dead device model would be worse than
// holding none.
- ObservationIndex.Clear();
- ConditionTracker.Clear();
- FanOut(SubscribedReferences(), ObservationIndex);
+ if (!TryCaptureSessionState(generation, out var restartIndex, out var restartTracker))
+ {
+ return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null);
+ }
+
+ // Not live while the index is empty (review m2): reads go to /current until the re-baseline lands.
+ _pumpLive = false;
+ restartIndex.Clear();
+ restartTracker.Clear();
+ FanOut(SubscribedReferences(), restartIndex);
// The driver's own state is consistent by this point, so it is safe to hand the
// restart to arbitrary caller code — which may call straight back into
@@ -2370,7 +2377,11 @@ public sealed partial class MTConnectDriver
.ConfigureAwait(false);
}
- ApplyAndFanOut(chunk);
+ if (!ApplyAndFanOut(chunk, generation))
+ {
+ return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null);
+ }
+
delivered = true;
// EVERY chunk advances the cursor, including an observation-free heartbeat: the
@@ -2379,9 +2390,12 @@ public sealed partial class MTConnectDriver
// like a gap.
expected = chunk.NextSequence;
- _streamDegraded = false;
- _pumpLive = true;
- PublishHealthy(DateTime.UtcNow);
+ // Review I-1: a pump wedged inside the fan-out above can come back after a restart replaced its
+ // session — its "the stream delivered" must then land nowhere.
+ if (!TryMarkStreamDelivered(generation))
+ {
+ return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null);
+ }
// R06:T-5 — a stream persistently behind the Agent reopens at the cursor it reached with a larger
// count, rather than falling out of the ring buffer and re-baselining every buffer-length.
@@ -2477,11 +2491,11 @@ public sealed partial class MTConnectDriver
}
PublishAgentCursor(generation, current.InstanceId, current.NextSequence);
- ApplyAndFanOut(current, baseline: true);
-
- _streamDegraded = false;
- _pumpLive = true;
- PublishHealthy(DateTime.UtcNow);
+ if (!ApplyAndFanOut(current, generation, baseline: true) || !TryMarkStreamDelivered(generation))
+ {
+ return new StreamOutcome(
+ StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, deliveredAny, null);
+ }
_logger.LogInformation(
"MTConnect driver {DriverInstanceId} re-baselined from {AgentUri}'s /current ({ObservationCount} observation(s), agent instanceId {InstanceId}) and resumes /sample from sequence {NextSequence}.",
@@ -2561,19 +2575,29 @@ public sealed partial class MTConnectDriver
/// Folds a /current snapshot or one /sample chunk into the shared index and
/// publishes every reference it touched to the handles that subscribe it.
///
- private void ApplyAndFanOut(MTConnectStreamsResult document, bool baseline = false)
+ /// The document to fold in.
+ /// The session generation of the pump doing it.
+ /// Whether it is a /current baseline.
+ /// (nothing applied) when the pump's session has been replaced.
+ private bool ApplyAndFanOut(MTConnectStreamsResult document, long generation, bool baseline = false)
{
- var index = ObservationIndex;
+ // Captured with the generation check under one lock (review m3): a stale chunk can never fold into the
+ // session state a restart installed after the pump's last check.
+ if (!TryCaptureSessionState(generation, out var index, out var tracker))
+ {
+ return false;
+ }
+
index.Apply(document, baseline);
// CONDITION observations also feed the alarm path, which keeps its own multi-condition state
// (a /sample Normal for one nativeCode is not a return to normal while another code is active).
- var touchedConditions = ConditionTracker.Apply(document, baseline);
+ var touchedConditions = tracker.Apply(document, baseline);
EmitConditionTransitions(touchedConditions, resync: false);
if (document.Observations is not { Count: > 0 })
{
- return;
+ return true;
}
// De-duplicated in document order: one Agent document may report a data item more than once
@@ -2602,6 +2626,74 @@ public sealed partial class MTConnectDriver
}
FanOut(touched, index);
+
+ return true;
+ }
+
+ ///
+ /// The observation index and condition tracker of the session names, read with
+ /// the generation check under — the lock installs
+ /// them, the session and the generation under. once that session is gone.
+ ///
+ private bool TryCaptureSessionState(
+ long generation, out MTConnectObservationIndex index, out MTConnectConditionTracker tracker)
+ {
+ lock (_subscriptionLock)
+ {
+ index = Volatile.Read(ref _index);
+ tracker = Volatile.Read(ref _conditions);
+
+ return _sessionGeneration == generation;
+ }
+ }
+
+ ///
+ /// The stream of the pump started under failed: degrade it — under the generation
+ /// check, in one critical section, so a superseded pump's failure never lands on the session that replaced it.
+ ///
+ /// The pump's session generation.
+ /// The failure (redacted text only).
+ /// when the pump's session has been replaced (nothing written).
+ private bool TryDegradeStream(long generation, string message)
+ {
+ // The mirror of TryMarkStreamDelivered: a superseded pump's failure is not the new session's.
+ lock (_subscriptionLock)
+ {
+ if (_sessionGeneration != generation)
+ {
+ return false;
+ }
+
+ _streamDegraded = true;
+ Degrade(message);
+
+ return true;
+ }
+ }
+
+ ///
+ /// Review I-1: the stream of the pump started under delivered — clear its
+ /// degradation, mark it live and publish Healthy, all under the generation check and in one critical section,
+ /// so a superseded pump (wedged in a subscriber past its last check) can never restore a replaced session's
+ /// health or make the read path trust a stream that is down. Nothing here calls caller code.
+ ///
+ /// The pump's session generation.
+ /// when the pump's session has been replaced (nothing written).
+ private bool TryMarkStreamDelivered(long generation)
+ {
+ lock (_subscriptionLock)
+ {
+ if (_sessionGeneration != generation)
+ {
+ return false;
+ }
+
+ _streamDegraded = false;
+ _pumpLive = true;
+ PublishHealthy(DateTime.UtcNow);
+
+ return true;
+ }
}
///
@@ -3045,24 +3137,27 @@ public sealed partial class MTConnectDriver
// The generation bump and the session write are ONE critical section: a Subscribe landing
// between them would start a pump on the OLD session stamped with the NEW generation, and
// the restart's own StartSampleStreamCore would then treat it as current.
+ var index = new MTConnectObservationIndex(binding.IndexTags);
+ index.Apply(current, baseline: true);
+
+ // the CONDITION tracker is per-session state like the index. Nothing is RAISED here — the
+ // lifecycle semaphore is held; the restarted pump delivers the diff as its first act.
+ var conditions = new MTConnectConditionTracker();
+ conditions.Apply(current, baseline: true);
+
+ // The index and the tracker are installed in the SAME critical section as the session and its
+ // generation (R06:T-1 / review m3): a pump that reads them under this lock with its generation still
+ // current gets this session's state, never the previous one's, and never the next one's.
lock (_subscriptionLock)
{
Volatile.Write(
ref _session,
new AgentSession(client, probe, current.InstanceId, current.NextSequence));
_sessionGeneration++;
+ Volatile.Write(ref _index, index);
+ Volatile.Write(ref _conditions, conditions);
}
- var index = new MTConnectObservationIndex(binding.IndexTags);
- index.Apply(current, baseline: true);
- Volatile.Write(ref _index, index);
-
- // the CONDITION tracker is per-session state like the index. Nothing is RAISED here — the
- // lifecycle semaphore is held; the restarted pump delivers the diff as its first act.
- var conditions = new MTConnectConditionTracker();
- conditions.Apply(current, baseline: true);
- Volatile.Write(ref _conditions, conditions);
-
// A fresh session has no failures behind it — including the /sample verdict, because a
// re-initialize is precisely the event that means an operator changed something.
_streamDegraded = false;
@@ -3134,8 +3229,16 @@ public sealed partial class MTConnectDriver
// Retire the whole session in one write, BEFORE disposing: a lock-free reader that still
// holds the old reference gets a disposed client (caught + named at each seam), never a
// session that is half-torn-down.
- var session = Volatile.Read(ref _session);
- Volatile.Write(ref _session, null);
+ // The retirement bumps the session generation too, so a pump abandoned by the teardown wait can no
+ // longer write health, flags or index state over what the shutdown / next start writes (review I-1).
+ AgentSession? session;
+ lock (_subscriptionLock)
+ {
+ session = Volatile.Read(ref _session);
+ Volatile.Write(ref _session, null);
+ _sessionGeneration++;
+ }
+
SafeDispose(session?.Client);
// A fresh index rather than Clear(): the tag table it coerces against belongs to the options
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
index 0f8ea7b8..b17d49cd 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStaleCursorTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStaleCursorTests.cs
@@ -107,4 +107,79 @@ public sealed class MTConnectStaleCursorTests
await driver.ShutdownAsync(CancellationToken.None);
}
}
+
+ ///
+ /// Review I-1: a pump wedged inside its fan-out that comes back after the restart must not restore the NEW
+ /// session's health (nor mark its stream live for the read path) while the new pump's stream is down. Positive
+ /// control: the CURRENT pump's own chunk still restores Healthy.
+ ///
+ [Fact]
+ public async Task An_abandoned_pump_does_not_restore_the_new_sessions_health()
+ {
+ var clientA = CannedAgentClient.FromFixtures();
+ var clientB = CannedAgentClient.FromFixtures();
+ clientB.SampleFailure = new IOException("connection refused");
+ var clients = new Queue([clientA, clientB]);
+
+ var driver = new MTConnectDriver(
+ new MTConnectDriverOptions
+ {
+ AgentUri = "http://fixture-agent:5000",
+ RequestTimeoutMs = 5000,
+ Reconnect = new MTConnectReconnectOptions { MinBackoffMs = 50, MaxBackoffMs = 200 },
+ Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
+ },
+ "mt-i1",
+ _ => 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);
+
+ await driver.InitializeAsync("{}", Ct).WaitAsync(Watchdog, Ct);
+
+ // The new session's pump fails its /sample: the driver goes Degraded.
+ await clientB.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
+ await WaitUntilAsync(() => driver.GetHealth().State == DriverState.Degraded, "the new stream is down");
+
+ wedge.Set();
+ await pumpA.WaitAsync(Watchdog, Ct);
+
+ driver.GetHealth().State.ShouldBe(
+ DriverState.Degraded, "a superseded pump must not report the new session's stream healthy");
+
+ // Positive control: the new session's own stream recovering does restore Healthy.
+ clientB.SampleFailure = null;
+ var calls = clientB.SampleCallCount;
+ await clientB.WaitForSampleCallsAsync(calls + 1).WaitAsync(Watchdog, Ct);
+ await clientB.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "2.5")).WaitAsync(Watchdog, Ct);
+ driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+ }
+ finally
+ {
+ wedge.Set();
+ await driver.ShutdownAsync(CancellationToken.None);
+ }
+ }
+
+ private static async Task WaitUntilAsync(Func condition, string because)
+ {
+ var deadline = DateTime.UtcNow + Watchdog;
+ while (!condition() && DateTime.UtcNow < deadline) await Task.Delay(20, Ct);
+ condition().ShouldBeTrue(because);
+ }
}