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

feat(mtconnect): ISubscribable /sample pump + ring-buffer re-baseline (Task 11)

Browse Source 
One shared /sample long poll behind every subscription handle. Subscribe fires
initial data per reference from the primed /current (Part 4 convention) and
returns without waiting on the stream; the pump runs on its own task under its
own token — never the caller's Subscribe token, which is disposed the moment
that call returns.

The pump owns every way the sequence can break:

- sequence gap, measured against the RUNNING cursor (the previous chunk's
  nextSequence). Comparing against the opening `from` reports a gap on every
  chunk once the buffer rolls — a /current re-baseline storm.
- OUT_OF_RANGE under HTTP 200 (InvalidDataException out of SampleAsync), which
  IsSequenceGap cannot see at all — without it a driver recovers from falling a
  little behind but not from falling a lot.
- agent restart, checked BEFORE the gap because a restart resets sequences and
  so usually trips the gap check too; it clears the index (the held values
  describe a device model that no longer exists) and tells the subscribers.
- every chunk advances the cursor, heartbeats included.

Re-baseline calls CurrentAsync directly on the client the pump already holds:
routing it through ReinitializeAsync would take the non-reentrant lifecycle
semaphore from inside a pump a lifecycle method may already be awaiting, and
hang the driver with no exception and no log. StopSampleStreamAsync is filled
in (same call sites) and InitializeAsync now stops the stream before teardown
too, so a re-Initialize on a live instance cannot leave a pump enumerating a
disposed client.

MTConnectStreamEnded/TimeoutException reconnect under a geometric backoff with
a 100 ms growth floor (MinBackoffMs defaults to 0, and 0 x multiplier is still
0). MTConnectStreamNotSupportedException does NOT retry — it latches, reports
loudly, and is cleared only by a re-initialize.

Health precedence is explicit: /current and /sample fail independently, so each
path owns a flag, clears only its own, and Healthy requires both clear. Neither
can paper over the other's failure.

29 new tests, all deterministic — no sleeps, no wall-clock assertions. The fake
grew per-generation sample streams (so a reconnect has somewhere to land),
scripted /current answers, scripted sample failures, and a chunk-consumed
signal that also fires when the consumer abandons the stream. 375/375 green.
This commit is contained in:
Joseph Doherty
2026-07-24 16:29:17 -04:00
parent 26eb511a7e
commit a41c26b5de
4 changed files with 1943 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs
+155 -10
View File
@@ -1,3 +1,4 @@
using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
@@ -34,9 +35,19 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
/// </remarks>
internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
{
private readonly Channel<ScriptedChunk> _chunks = Channel.CreateUnbounded<ScriptedChunk>();
private readonly CancellationTokenSource _disposeCts = new();
/// <summary>Chunks a test has scripted but not yet pumped — see <see cref="ScriptChunks"/>.</summary>
private readonly ConcurrentQueue<MTConnectStreamsResult> _script = new();
/// <summary>
/// The <b>current</b> <c>/sample</c> stream generation. Replaced (not merely completed) by
/// <see cref="EndStream"/> so that a pump which reconnects after a dropped stream gets a
/// live channel to read from instead of one that is permanently closed — without that, a
/// reconnect test could only ever observe the reconnect failing.
/// </summary>
private Channel<ScriptedChunk> _chunks = Channel.CreateUnbounded<ScriptedChunk>();
private int _probeCallCount;
private int _currentCallCount;
private int _sampleCallCount;
@@ -66,6 +77,44 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
public static MTConnectStreamsResult Chunk(string fixture) =>
MTConnectStreamsParser.Parse(File.ReadAllText(fixture));
/// <summary>
/// An Agent scripted for the ring-buffer-overflow story: the driver primes from
/// <c>Fixtures/current.xml</c> (<c>nextSequence</c> 108), then the first scripted chunk is
/// <c>Fixtures/sample-gap.xml</c> — whose <c>firstSequence</c> (5000) is strictly newer than
/// the cursor, i.e. the buffer rolled past the driver — and the second is an ordinary chunk
/// that continues contiguously from the re-baseline.
/// </summary>
/// <remarks>
/// The <b>second</b> <c>/current</c> answer is what makes this a story rather than a single
/// event: it advertises the buffer the gap chunk revealed (<c>firstSequence</c> 5000 /
/// <c>nextSequence</c> 5005), so a driver that re-baselines and resumes from that answer sees
/// the follow-up chunk as contiguous. A driver that re-baselined but resumed from the stale
/// cursor would report a gap on every subsequent chunk instead.
/// </remarks>
public static CannedAgentClient WithGapThenResume()
{
var client = FromFixtures();
var gap = Chunk("Fixtures/sample-gap.xml");
// 1st /current = the InitializeAsync prime (the fixture). 2nd = the post-gap re-baseline.
client.CurrentAnswers.Enqueue(client.Current);
client.CurrentAnswers.Enqueue(
new MTConnectStreamsResult(gap.InstanceId, gap.NextSequence, gap.FirstSequence, gap.Observations));
client.ScriptChunks(
gap,
new MTConnectStreamsResult(
gap.InstanceId,
gap.NextSequence + 5,
gap.NextSequence,
[
new MTConnectObservation(
"dev1_pos", "201.5000", new DateTime(2026, 7, 24, 12, 6, 0, DateTimeKind.Utc)),
]));
return client;
}
/// <summary>The document <c>/probe</c> answers with. Settable so a test can re-shape the model.</summary>
public MTConnectProbeModel Probe { get; set; }
@@ -81,6 +130,31 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
/// <summary>When set, <c>/current</c> throws this instead of answering.</summary>
public Exception? CurrentFailure { get; set; }
/// <summary>
/// Scripted <c>/current</c> answers, consumed in order: each call dequeues the next one and
/// <b>promotes it to <see cref="Current"/></b>, so once the script runs out the Agent keeps
/// answering with the last thing it said rather than travelling back in time. This is how a
/// test scripts "the prime, then the post-gap re-baseline" without racing the pump for a
/// property setter.
/// </summary>
public ConcurrentQueue<MTConnectStreamsResult> CurrentAnswers { get; } = new();
/// <summary>
/// Scripted one-shot <c>/sample</c> failures: each enumeration dequeues one and throws it.
/// Models a transient stream fault the Agent recovers from — e.g. the
/// <see cref="InvalidDataException"/> a real cppagent's <c>OUT_OF_RANGE</c> error document
/// (served under HTTP 200) surfaces as.
/// </summary>
public ConcurrentQueue<Exception> SampleFailures { get; } = new();
/// <summary>
/// A <b>sticky</b> <c>/sample</c> failure: every enumeration throws it, forever. Models a
/// configuration-level fault such as
/// <see cref="MTConnectStreamNotSupportedException"/>, where reconnecting reproduces the
/// identical answer — the shape a pump must NOT retry-loop against.
/// </summary>
public Exception? SampleFailure { get; set; }
/// <summary>
/// When set, the <b>next</b> <c>/probe</c> parks here until the test completes it, then the
/// gate clears itself so later calls answer immediately.
@@ -101,6 +175,14 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
/// </summary>
public TaskCompletionSource? CurrentGate { get; set; }
/// <summary>
/// Signalled the moment a <c>/current</c> request <b>lands</b> — before
/// <see cref="CurrentGate"/> parks it. This is the deterministic "the caller is now inside
/// the request" barrier a test needs before landing a concurrent lifecycle change on it;
/// without it the only alternative is polling <see cref="CurrentCallCount"/> on a timer.
/// </summary>
public TaskCompletionSource? CurrentEntered { get; set; }
/// <summary>Number of <c>/probe</c> requests issued against this client.</summary>
public int ProbeCallCount => Volatile.Read(ref _probeCallCount);
@@ -152,6 +234,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
ObjectDisposedException.ThrowIf(IsDisposed, this);
ct.ThrowIfCancellationRequested();
Interlocked.Increment(ref _currentCallCount);
CurrentEntered?.TrySetResult();
// The request was accepted; the answer is withheld until the test releases the gate or the
// caller's deadline cancels the token.
@@ -161,6 +244,12 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
await gate.Task.WaitAsync(ct).ConfigureAwait(false);
}
// A scripted answer supersedes — and then becomes — the standing one.
if (CurrentAnswers.TryDequeue(out var scripted))
{
Current = scripted;
}
return CurrentFailure is null ? Current : throw CurrentFailure;
}
@@ -172,6 +261,21 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
Interlocked.Increment(ref _sampleCallCount);
LastSampleFrom = from;
// Thrown before the first chunk, exactly like a real client that fails its /sample request:
// one-shot scripts first, then the sticky "this endpoint will never stream" failure.
if (SampleFailures.TryDequeue(out var oneShot))
{
throw oneShot;
}
if (SampleFailure is { } sticky)
{
throw sticky;
}
// Bound to THIS generation: EndStream swaps in a fresh channel, so a consumer that
// reconnects reads the new one rather than the closed one it just lost.
var chunks = Volatile.Read(ref _chunks);
using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(ct, _disposeCts.Token);
var delivered = 0L;
@@ -180,7 +284,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
ScriptedChunk scripted;
try
{
scripted = await _chunks.Reader.ReadAsync(lifetime.Token).ConfigureAwait(false);
scripted = await chunks.Reader.ReadAsync(lifetime.Token).ConfigureAwait(false);
}
catch (ChannelClosedException)
{
@@ -192,9 +296,18 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
delivered++;
yield return scripted.Result;
scripted.Consumed.TrySetResult();
try
{
yield return scripted.Result;
}
finally
{
// Signalled when the consumer is DONE with this chunk — whether it came back for the
// next one or abandoned the enumeration. The finally is load-bearing: a chunk that
// makes the pump break out (a sequence gap, an agent restart) is never resumed past,
// so signalling only after the resume would hang every re-baseline test.
scripted.Consumed.TrySetResult();
}
}
}
@@ -212,7 +325,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
var scripted = new ScriptedChunk(
chunk, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
if (!_chunks.Writer.TryWrite(scripted))
if (!Volatile.Read(ref _chunks).Writer.TryWrite(scripted))
{
throw new InvalidOperationException("The scripted /sample stream is already closed.");
}
@@ -221,10 +334,42 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
}
/// <summary>
/// Closes the scripted <c>/sample</c> stream, which surfaces to the consumer as an
/// <see cref="MTConnectStreamEndedException"/> — the transient, reconnectable end.
/// Queues chunks for <see cref="PumpOnce"/> to deliver one at a time. Nothing is sent until
/// a test asks for it — the script is a plan, not a schedule.
/// </summary>
public void EndStream() => _chunks.Writer.TryComplete();
/// <param name="chunks">The chunks the Agent should send, in order.</param>
public void ScriptChunks(params MTConnectStreamsResult[] chunks)
{
ArgumentNullException.ThrowIfNull(chunks);
foreach (var chunk in chunks)
{
_script.Enqueue(chunk);
}
}
/// <summary>
/// Pumps the next scripted chunk (see <see cref="ScriptChunks"/>) and waits until the
/// consumer has finished with it.
/// </summary>
/// <returns>A task completing once the consumer has processed — or abandoned the stream on — that chunk.</returns>
/// <exception cref="InvalidOperationException">The script is exhausted.</exception>
public Task PumpOnce() =>
_script.TryDequeue(out var next)
? PumpAsync(next)
: throw new InvalidOperationException(
"No scripted /sample chunk left to pump; call ScriptChunks first.");
/// <summary>
/// Closes the current scripted <c>/sample</c> stream, which surfaces to its consumer as an
/// <see cref="MTConnectStreamEndedException"/> — the transient, reconnectable end — and opens
/// a fresh one so a reconnecting consumer has somewhere to land.
/// </summary>
public void EndStream()
{
var closed = Interlocked.Exchange(ref _chunks, Channel.CreateUnbounded<ScriptedChunk>());
closed.Writer.TryComplete();
}
/// <inheritdoc/>
public void Dispose()
@@ -237,7 +382,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
}
_disposeCts.Cancel();
_chunks.Writer.TryComplete();
Volatile.Read(ref _chunks).Writer.TryComplete();
_disposeCts.Dispose();
}