dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(mtconnect): a superseded pump writes no health, flags or index state (review I-1, m2, m3)

Browse Source 
- The pump's delivered/degraded writes (stream flag, live flag, PublishHealthy /
  Degrade) now happen in TryMarkStreamDelivered / TryDegradeStream: one critical
  section under _subscriptionLock that re-checks the session generation. A pump
  wedged in a subscriber past its last check can no longer report a replaced
  session's (down) stream Healthy and make the read path trust it.
- m3: the index and condition tracker are installed in the same lock as the
  session + generation and captured under it (TryCaptureSessionState), so a stale
  chunk cannot fold into the new session's index.
- Teardown also bumps the generation, so an abandoned pump cannot write over the
  shutdown's health either.
- m2: an Agent restart clears the live flag before clearing the index, so reads
  go to /current until the re-baseline lands.
Red test: the wedged-pump harness with the new stream failing, plus a positive
control (the current pump's chunk restores Healthy).

Refs #743

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 16:27:23 -04:00
parent b943f48e01
commit f48a065c21
2 changed files with 215 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
+140 -37
View File
@@ -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 <c>/current</c> snapshot or one <c>/sample</c> chunk into the shared index and
/// publishes every reference it touched to the handles that subscribe it.
/// </summary>
private void ApplyAndFanOut(MTConnectStreamsResult document, bool baseline = false)
/// <param name="document">The document to fold in.</param>
/// <param name="generation">The session generation of the pump doing it.</param>
/// <param name="baseline">Whether it is a <c>/current</c> baseline.</param>
/// <returns><see langword="false"/> (nothing applied) when the pump's session has been replaced.</returns>
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;
}
/// <summary>
/// The observation index and condition tracker of the session <paramref name="generation"/> names, read with
/// the generation check under <see cref="_subscriptionLock"/> — the lock <see cref="StartCoreAsync"/> installs
/// them, the session and the generation under. <see langword="false"/> once that session is gone.
/// </summary>
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;
}
}
/// <summary>
/// The stream of the pump started under <paramref name="generation"/> 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.
/// </summary>
/// <param name="generation">The pump's session generation.</param>
/// <param name="message">The failure (redacted text only).</param>
/// <returns><see langword="false"/> when the pump's session has been replaced (nothing written).</returns>
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;
}
}
/// <summary>
/// Review I-1: the stream of the pump started under <paramref name="generation"/> 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.
/// </summary>
/// <param name="generation">The pump's session generation.</param>
/// <returns><see langword="false"/> when the pump's session has been replaced (nothing written).</returns>
private bool TryMarkStreamDelivered(long generation)
{
lock (_subscriptionLock)
{
if (_sessionGeneration != generation)
{
return false;
}
_streamDegraded = false;
_pumpLive = true;
PublishHealthy(DateTime.UtcNow);
return true;
}
}
/// <summary>
@@ -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
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStaleCursorTests.cs
+75
View File
@@ -107,4 +107,79 @@ public sealed class MTConnectStaleCursorTests
await driver.ShutdownAsync(CancellationToken.None);
}
}
/// <summary>
/// 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.
/// </summary>
[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<CannedAgentClient>([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<bool> condition, string because)
{
var deadline = DateTime.UtcNow + Watchdog;
while (!condition() && DateTime.UtcNow < deadline) await Task.Delay(20, Ct);
condition().ShouldBeTrue(because);
}
}