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 8dcc9f68..36214f5a 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs
@@ -42,19 +42,19 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// nextSequence.
///
///
-/// Scope. This type currently implements and
-/// . ISubscribable (Task 11), ITagDiscovery (Task 12)
+/// Scope. This type currently implements ,
+/// and . ITagDiscovery (Task 12)
/// and IHostConnectivityProbe/IRediscoverable (Task 13) are added on top of
/// the state this class already caches — the probe model, the Agent instanceId, and
/// the observation index.
///
///
/// The read path never writes the shared observation index — see
-/// . That index has exactly one writer, and Task 11's /sample
-/// pump is it.
+/// . That index has exactly one writer: the /sample pump
+/// ().
///
///
-public sealed class MTConnectDriver : IDriver, IReadable
+public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable
{
///
/// Coarse per-entry byte weights behind . The contract asks
@@ -85,6 +85,19 @@ public sealed class MTConnectDriver : IDriver, IReadable
///
private const uint BadNotConnected = 0x808A0000u;
+ ///
+ /// Floor the reconnect backoff grows from, in milliseconds. Load-bearing: the shipped
+ /// default is 0 (an immediate
+ /// first retry, which is what an operator wants after a one-off blip) and
+ /// 0 × multiplier is still 0 — so without a floor, the "geometric" backoff would be
+ /// an unbounded hot loop against an Agent that stays down. Mirrors
+ /// ModbusTcpTransport.ConnectWithBackoffAsync.
+ ///
+ private const int BackoffGrowthFloorMs = 100;
+
+ /// Growth factor used when the authored multiplier could not grow the delay (≤ 1).
+ private const double DefaultBackoffMultiplier = 2.0;
+
///
/// Config-JSON reader options, mirroring the sibling driver factories. Note there is
/// deliberately no JsonStringEnumConverter: enum-carrying DTO fields stay
@@ -145,10 +158,58 @@ public sealed class MTConnectDriver : IDriver, IReadable
///
private readonly SemaphoreSlim _lifecycle = new(1, 1);
+ ///
+ /// Guards the subscription registry and the pump's control block (,
+ /// , ).
+ ///
+ ///
+ ///
+ /// Lock order is → this, never the reverse. The lifecycle
+ /// methods take this lock (to start/stop the pump) while holding the semaphore;
+ /// therefore releases this lock before awaiting
+ /// , and nothing under this lock ever waits on
+ /// .
+ ///
+ ///
+ /// Nothing is awaited while it is held, and no subscriber callback is raised under
+ /// it. A handler is caller code: it may block, throw, or re-enter this driver, and
+ /// holding a lock across it would turn a slow subscriber into a stalled pump.
+ ///
+ ///
+ private readonly Lock _subscriptionLock = new();
+
+ /// Live subscriptions by id. Guarded by .
+ private readonly Dictionary _subscriptions = [];
+
private MTConnectDriverOptions _options;
private MTConnectObservationIndex _index;
- private long? _agentInstanceId;
private DriverHealth _health = new(DriverState.Unknown, null, null);
+ private long _nextSubscriptionId;
+
+ /// Cancels the shared /sample pump. Guarded by .
+ private CancellationTokenSource? _pumpCts;
+
+ /// The shared /sample pump task. Guarded by .
+ private Task? _pumpTask;
+
+ ///
+ /// 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
+ /// later is not new information. Cleared only by a lifecycle
+ /// start, which is the one event that means an operator changed something. Guarded by
+ /// .
+ ///
+ private bool _streamUnsupported;
+
+ ///
+ /// Whether the /sample pump is currently failing. Kept separate from
+ /// because /sample and /current are different
+ /// Agent requests that fail independently — see .
+ ///
+ private volatile bool _streamDegraded;
+
+ /// Whether the /current read path is currently failing. See .
+ private volatile bool _readDegraded;
///
/// The live agent client plus the /probe cache derived from it, held as one
@@ -247,10 +308,28 @@ public sealed class MTConnectDriver : IDriver, IReadable
internal MTConnectProbeModel? CachedProbeModel => Volatile.Read(ref _session)?.ProbeModel;
///
- /// The Agent's instanceId as of the last successful /current. Task 13 watches
- /// this for change to raise rediscovery (an Agent restart invalidates every cached id).
+ /// The Agent's instanceId as of the last successful /current — including the
+ /// re-baseline the /sample pump runs when it sees the id change mid-stream. Task 13
+ /// watches this for change to raise rediscovery (an Agent restart invalidates every cached
+ /// id).
///
- internal long? AgentInstanceId => _agentInstanceId;
+ internal long? AgentInstanceId => Volatile.Read(ref _session)?.InstanceId;
+
+ ///
+ /// The shared /sample pump's task while one is running, else null. Exposed to
+ /// tests as the deterministic teardown barrier: "the pump has stopped" is a task completion,
+ /// never a sleep. Guarded, so a test never observes a half-installed pump.
+ ///
+ internal Task? SampleStreamTask
+ {
+ get
+ {
+ lock (_subscriptionLock)
+ {
+ return _pumpTask;
+ }
+ }
+ }
/// The options currently in force — the constructor's, or the last document applied.
internal MTConnectDriverOptions EffectiveOptions => _options;
@@ -275,7 +354,12 @@ public sealed class MTConnectDriver : IDriver, IReadable
// be the exact outcome ReinitializeAsync is written to avoid.
var options = ResolveIncomingOptions(driverConfigJson);
- // A re-Initialize on a live instance must not orphan the previous client.
+ // A re-Initialize on a live instance must not orphan the previous client — nor the pump
+ // that is enumerating it. Stopping the stream FIRST is the same ordering
+ // ReinitializeAsync and ShutdownAsync use, and for the same reason: a pump left
+ // enumerating a disposed client reads the disposal as a dropped connection and
+ // reconnects against an object that no longer exists.
+ await StopSampleStreamAsync().ConfigureAwait(false);
await TeardownCoreAsync().ConfigureAwait(false);
await StartCoreAsync(options, existingClient: null, cancellationToken).ConfigureAwait(false);
@@ -504,7 +588,8 @@ public sealed class MTConnectDriver : IDriver, IReadable
results[i] = snapshot.Get(fullReferences[i]);
}
- WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
+ _readDegraded = false;
+ PublishHealthy(DateTime.UtcNow);
return results;
}
@@ -523,6 +608,648 @@ public sealed class MTConnectDriver : IDriver, IReadable
}
}
+ // ---- ISubscribable: ONE shared /sample long poll behind every handle ----
+
+ ///
+ public event EventHandler? OnDataChange;
+
+ ///
+ ///
+ ///
+ /// Initial data comes from the primed /current, not from the wire. Every
+ /// subscribed reference gets one callback immediately (the OPC UA Part 4 convention),
+ /// answered out of the observation index — including the references with no value yet,
+ /// which report BadWaitingForInitialData. Waiting for the Agent's next chunk
+ /// instead would leave a fresh subscription silent for up to a whole heartbeat,
+ /// and permanently silent for any data item that simply is not changing.
+ ///
+ ///
+ /// Subscribe does not wait for the stream. The first subscription starts the
+ /// shared pump; the pump then opens /sample on its own task under its own
+ /// cancellation token. This is deliberate on both counts. The opening handshake cannot
+ /// be awaited inside the caller's Subscribe timeout in any useful sense — the first
+ /// chunk may legitimately be a heartbeat away — and the long-lived poll must NOT run
+ /// under the caller's token, which belongs to one Subscribe call and is disposed the
+ /// moment it returns. What bounds the running stream is the client's own heartbeat
+ /// watchdog, and what stops it is or the lifecycle.
+ ///
+ ///
+ /// is not honoured per subscription, and
+ /// cannot be: MTConnect's publish cadence is the Agent-side interval query
+ /// parameter of the single shared /sample request
+ /// (), fixed when the client is
+ /// built. Honouring a per-handle interval would mean one stream per subscription — the
+ /// Agent load this driver exists to avoid — or client-side throttling that would drop
+ /// transient EVENT values the Agent bothered to send. The parameter is accepted and
+ /// ignored, which is also what the natively-subscribing sibling drivers do.
+ ///
+ ///
+ /// Subscribing before is legal: the interest is recorded,
+ /// every reference reports "no value yet", and the initialize that follows starts the
+ /// pump for it. It does not throw and it does not dial an Agent that is not there yet.
+ ///
+ ///
+ ///
+ /// is null — a caller bug, and the only thing this
+ /// method is loud about, matching 's posture.
+ ///
+ public Task SubscribeAsync(
+ IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
+ {
+ ArgumentNullException.ThrowIfNull(fullReferences);
+
+ // De-duplicated: a reference asked for twice in one subscription is one subscribed value,
+ // and would otherwise raise two identical callbacks per chunk forever.
+ var references = new HashSet(StringComparer.Ordinal);
+ foreach (var reference in fullReferences)
+ {
+ if (reference is not null)
+ {
+ references.Add(reference);
+ }
+ }
+
+ var handle = new MTConnectSampleHandle(Interlocked.Increment(ref _nextSubscriptionId));
+
+ lock (_subscriptionLock)
+ {
+ _subscriptions[handle.SubscriptionId] = new SubscriptionState(handle, references);
+
+ // No-op when a pump is already running: every handle shares the one stream.
+ StartSampleStreamCore(republishOnStart: false);
+ }
+
+ // Raised OUTSIDE the lock — a handler is caller code (see the _subscriptionLock remarks).
+ var index = ObservationIndex;
+ foreach (var reference in references)
+ {
+ RaiseDataChange(handle, reference, index.Get(reference));
+ }
+
+ _logger.LogDebug(
+ "MTConnect driver {DriverInstanceId} opened subscription {SubscriptionId} over {ReferenceCount} reference(s).",
+ _driverInstanceId, handle.DiagnosticId, references.Count);
+
+ return Task.FromResult(handle);
+ }
+
+ ///
+ ///
+ /// Drops the handle's references and stops the shared stream when the last subscription
+ /// goes. An unknown handle — already unsubscribed, or issued by a different driver — is a
+ /// no-op rather than an error: this runs on teardown paths, including failure paths, where
+ /// throwing would mask whatever prompted the teardown.
+ ///
+ /// is null.
+ public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
+ {
+ ArgumentNullException.ThrowIfNull(handle);
+
+ if (handle is not MTConnectSampleHandle ours)
+ {
+ return;
+ }
+
+ bool stopStream;
+ lock (_subscriptionLock)
+ {
+ if (!_subscriptions.Remove(ours.SubscriptionId))
+ {
+ return;
+ }
+
+ stopStream = !HasSubscribedReferencesCore();
+ }
+
+ if (stopStream)
+ {
+ // Outside the lock: StopSampleStreamAsync awaits the pump, and the pump takes this same
+ // lock to read the subscription registry. Holding it here would deadlock the two.
+ await StopSampleStreamAsync().ConfigureAwait(false);
+ }
+
+ _logger.LogDebug(
+ "MTConnect driver {DriverInstanceId} closed subscription {SubscriptionId}{StreamNote}.",
+ _driverInstanceId, ours.DiagnosticId, stopStream ? " and stopped the shared /sample stream" : string.Empty);
+ }
+
+ ///
+ /// Starts the shared /sample pump, if there is anything to pump and anything to pump
+ /// from. Caller must hold .
+ ///
+ ///
+ /// Whether the pump should republish the observation index to every subscriber before it
+ /// opens the stream — true when a lifecycle start replaced the baseline the subscribers'
+ /// current values came from.
+ ///
+ private void StartSampleStreamCore(bool republishOnStart)
+ {
+ if (_pumpTask is not null || _streamUnsupported || !HasSubscribedReferencesCore())
+ {
+ return;
+ }
+
+ // One read of the session: the client the pump enumerates and the cursor it opens at are
+ // guaranteed to belong to the same initialize.
+ var session = Volatile.Read(ref _session);
+ if (session is null)
+ {
+ return;
+ }
+
+ var options = _options;
+ var cts = new CancellationTokenSource();
+ _pumpCts = cts;
+ _pumpTask = Task.Run(
+ () => RunSampleStreamAsync(
+ session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, cts.Token),
+ CancellationToken.None);
+ }
+
+ ///
+ /// The shared /sample pump: reads chunks, keeps the observation index and the
+ /// subscribers up to date, and owns every way the stream can stop.
+ ///
+ ///
+ ///
+ /// Never throws. It is a detached background task; an escaping exception would be
+ /// an unobserved fault that silently ends the driver's subscription surface while every
+ /// handle still looks alive.
+ ///
+ ///
+ /// Never calls a lifecycle method. Its re-baseline goes straight to
+ /// on the client it already holds — see
+ /// and the remarks for why
+ /// "just re-initialize" would be a silent permanent deadlock.
+ ///
+ ///
+ /// Backoff applies to failures only. A re-baseline is a protocol event, not a
+ /// fault, so it reopens the stream immediately and resets the attempt count; only a
+ /// genuine failure (dropped connection, blown heartbeat, a re-baseline that itself
+ /// failed) advances .
+ ///
+ ///
+ private async Task RunSampleStreamAsync(
+ IMTConnectAgentClient client,
+ MTConnectDriverOptions options,
+ long from,
+ long instanceId,
+ bool republishOnStart,
+ CancellationToken ct)
+ {
+ var cursor = from;
+ var attempt = 0;
+
+ try
+ {
+ if (republishOnStart)
+ {
+ FanOut(SubscribedReferences(), ObservationIndex);
+ }
+
+ while (!ct.IsCancellationRequested)
+ {
+ if (attempt > 0)
+ {
+ var delay = BackoffFor(attempt, options.Reconnect);
+ if (delay > TimeSpan.Zero)
+ {
+ try
+ {
+ await Task.Delay(delay, ct).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException)
+ {
+ return;
+ }
+ }
+ }
+
+ var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, ct).ConfigureAwait(false);
+ cursor = outcome.Cursor;
+ instanceId = outcome.InstanceId;
+
+ switch (outcome.Verdict)
+ {
+ case StreamVerdict.Cancelled:
+ return;
+
+ case StreamVerdict.Unsupported:
+ // Configuration, not weather: reconnecting reproduces the identical answer
+ // forever, so retrying would be an unbounded hot loop against an endpoint
+ // that will never stream until an operator changes something. Latch it, say
+ // so loudly, and stop. A re-initialize clears the latch.
+ lock (_subscriptionLock)
+ {
+ _streamUnsupported = true;
+ }
+
+ _streamDegraded = true;
+ Degrade(outcome.Error!);
+
+ _logger.LogError(
+ outcome.Error,
+ "MTConnect driver {DriverInstanceId} cannot stream /sample from {AgentUri}: the endpoint did not answer with a framed multipart stream. Subscriptions stay registered but will receive no updates until the driver is re-initialized; check for a proxy or load balancer in front of the Agent.",
+ _driverInstanceId, options.AgentUri);
+
+ return;
+
+ case StreamVerdict.Reopen:
+ attempt = 0;
+
+ break;
+
+ default:
+ _streamDegraded = true;
+ Degrade(outcome.Error!);
+ attempt++;
+
+ _logger.LogWarning(
+ outcome.Error,
+ "MTConnect driver {DriverInstanceId} lost the /sample stream from {AgentUri}; reconnecting from sequence {Cursor} (attempt {Attempt}).",
+ _driverInstanceId, options.AgentUri, cursor, attempt);
+
+ break;
+ }
+ }
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(
+ ex,
+ "MTConnect driver {DriverInstanceId} stopped its /sample pump on an unexpected fault; subscriptions will receive no further updates until the driver is re-initialized.",
+ _driverInstanceId);
+
+ _streamDegraded = true;
+ Degrade(ex);
+ }
+ }
+
+ ///
+ /// Enumerates one /sample connection until it ends, and classifies how it ended.
+ /// Re-baselines are performed here, inside the enumeration, so the index and the
+ /// subscribers are up to date before the stream is reopened.
+ ///
+ private async Task ConsumeStreamAsync(
+ IMTConnectAgentClient client,
+ MTConnectDriverOptions options,
+ long cursor,
+ long instanceId,
+ CancellationToken ct)
+ {
+ // The RUNNING cursor the gap check is made against: the previous chunk's nextSequence, equal
+ // to the opening `from` only for the first chunk. Comparing every chunk against the opening
+ // `from` instead reports a gap on every chunk once the Agent's buffer rolls past it — an
+ // endless /current re-baseline storm against a perfectly healthy stream.
+ var expected = cursor;
+
+ try
+ {
+ await foreach (var chunk in client.SampleAsync(cursor, ct).ConfigureAwait(false))
+ {
+ if (chunk is null)
+ {
+ continue;
+ }
+
+ // 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);
+ // a restart invalidates every one of them, because the device model they describe
+ // no longer exists. Testing the gap first would misdiagnose the restart and keep
+ // serving values the new Agent may never report again.
+ if (chunk.InstanceId != instanceId)
+ {
+ _logger.LogWarning(
+ "MTConnect driver {DriverInstanceId} saw {AgentUri} restart (instanceId {Previous} -> {Current}); dropping every held observation and re-baselining.",
+ _driverInstanceId, options.AgentUri, instanceId, chunk.InstanceId);
+
+ // 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();
+ FanOut(SubscribedReferences(), ObservationIndex);
+
+ return await RebaselineAsync(client, options, expected, instanceId, ct).ConfigureAwait(false);
+ }
+
+ if (IMTConnectAgentClient.IsSequenceGap(expected, chunk))
+ {
+ _logger.LogWarning(
+ "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, ct).ConfigureAwait(false);
+ }
+
+ ApplyAndFanOut(chunk);
+
+ // EVERY chunk advances the cursor, including an observation-free heartbeat: the
+ // Agent sends those precisely so a quiet connection can be told from a dead one, and
+ // they carry the sequence forward. Not advancing on one makes the NEXT chunk look
+ // like a gap.
+ expected = chunk.NextSequence;
+
+ _streamDegraded = false;
+ PublishHealthy(DateTime.UtcNow);
+ }
+
+ // The seam contracts that cancellation is the ONLY way this enumeration ends quietly, so
+ // a quiet end under a live token is a non-conforming client — reported as the lost
+ // stream it is rather than treated as "the subscription is simply idle" (#485).
+ return ct.IsCancellationRequested
+ ? new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, null)
+ : new StreamOutcome(
+ StreamVerdict.Transient,
+ expected,
+ instanceId,
+ new MTConnectStreamEndedException(
+ MTConnectStreamEndReason.ConnectionClosed, options.AgentUri, 0));
+ }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested)
+ {
+ return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, null);
+ }
+ catch (MTConnectStreamNotSupportedException ex)
+ {
+ return new StreamOutcome(StreamVerdict.Unsupported, expected, instanceId, ex);
+ }
+ catch (InvalidDataException ex)
+ {
+ // THE OTHER HALF OF RING-BUFFER OVERFLOW, and the one IsSequenceGap cannot see: a real
+ // cppagent answers a `from` that has already fallen out of its buffer with an
+ // MTConnectError / OUT_OF_RANGE document served under HTTP 200, which the client
+ // surfaces here rather than as a gap-bearing chunk. Re-baselining on the gap alone would
+ // mean the driver recovers from falling a little behind but not from falling a lot.
+ _logger.LogWarning(
+ ex,
+ "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, ct).ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ // Transient by default: a dropped connection, a closing boundary, the heartbeat
+ // watchdog's TimeoutException. Reconnect under backoff.
+ return new StreamOutcome(StreamVerdict.Transient, expected, instanceId, ex);
+ }
+ }
+
+ ///
+ /// Re-primes the observation index from a fresh /current and reports where the stream
+ /// should resume. The recovery path for every way the incremental sequence can be broken:
+ /// a ring-buffer overflow (detected as a gap, or refused outright as OUT_OF_RANGE)
+ /// and an Agent restart.
+ ///
+ ///
+ /// This calls the client directly and takes no lock. Re-baselining by calling
+ /// — the obvious way to write it, since that method already
+ /// does exactly this work — would take the non-reentrant semaphore
+ /// from inside a pump that a lifecycle method may already be waiting on, and hang the driver
+ /// permanently with no exception and no log line.
+ ///
+ private async Task RebaselineAsync(
+ IMTConnectAgentClient client,
+ MTConnectDriverOptions options,
+ long fallbackCursor,
+ long fallbackInstanceId,
+ CancellationToken ct)
+ {
+ try
+ {
+ var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, ct)
+ .ConfigureAwait(false);
+
+ PublishAgentInstanceId(current.InstanceId);
+ ApplyAndFanOut(current);
+
+ _streamDegraded = false;
+ PublishHealthy(DateTime.UtcNow);
+
+ _logger.LogInformation(
+ "MTConnect driver {DriverInstanceId} re-baselined from {AgentUri}'s /current ({ObservationCount} observation(s), agent instanceId {InstanceId}) and resumes /sample from sequence {NextSequence}.",
+ _driverInstanceId,
+ options.AgentUri,
+ current.Observations.Count,
+ current.InstanceId,
+ current.NextSequence);
+
+ return new StreamOutcome(StreamVerdict.Reopen, current.NextSequence, current.InstanceId, null);
+ }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested)
+ {
+ return new StreamOutcome(StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, null);
+ }
+ catch (Exception ex)
+ {
+ // The Agent is unreachable or unusable right now. Treated as an ordinary transient
+ // failure so the reconnect backoff applies — without it, an Agent that is down would be
+ // re-baselined against as fast as the loop can spin. The cursor is deliberately left
+ // where it was: reopening there either works or lands right back here.
+ return new StreamOutcome(StreamVerdict.Transient, fallbackCursor, fallbackInstanceId, ex);
+ }
+ }
+
+ ///
+ /// The reconnect delay before attempt , as a pure function — the
+ /// backoff policy is asserted directly rather than inferred from how long a test slept.
+ ///
+ ///
+ /// The first retry honours (0 by
+ /// default = immediate, which is what an operator wants after a one-off blip); every later
+ /// one multiplies, from a floor, up to
+ /// . Operator-authored nonsense cannot
+ /// un-bound the loop: a multiplier that cannot grow the delay falls back to
+ /// , and negative values clamp to zero.
+ ///
+ /// The 1-based reconnect attempt about to be made.
+ /// The authored backoff options.
+ /// How long to wait before that attempt.
+ internal static TimeSpan BackoffFor(int attempt, MTConnectReconnectOptions reconnect)
+ {
+ ArgumentNullException.ThrowIfNull(reconnect);
+
+ var capMs = Math.Max(0, reconnect.MaxBackoffMs);
+ var minMs = Math.Max(0, reconnect.MinBackoffMs);
+
+ if (attempt <= 1)
+ {
+ return TimeSpan.FromMilliseconds(Math.Min(minMs, capMs));
+ }
+
+ var multiplier = reconnect.BackoffMultiplier > 1.0 ? reconnect.BackoffMultiplier : DefaultBackoffMultiplier;
+ var delayMs = (double)Math.Max(minMs, BackoffGrowthFloorMs);
+
+ for (var step = 2; step <= attempt; step++)
+ {
+ delayMs *= multiplier;
+
+ if (delayMs >= capMs)
+ {
+ return TimeSpan.FromMilliseconds(capMs);
+ }
+ }
+
+ return TimeSpan.FromMilliseconds(Math.Min(delayMs, capMs));
+ }
+
+ ///
+ /// 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)
+ {
+ var index = ObservationIndex;
+ index.Apply(document);
+
+ if (document.Observations is not { Count: > 0 })
+ {
+ return;
+ }
+
+ // De-duplicated in document order: one Agent document may report a data item more than once
+ // (a Condition container carries several simultaneously-active states), and the index has
+ // already reconciled those into a single snapshot.
+ var touched = new List(document.Observations.Count);
+ var seen = new HashSet(StringComparer.Ordinal);
+ foreach (var observation in document.Observations)
+ {
+ if (observation is null || string.IsNullOrWhiteSpace(observation.DataItemId))
+ {
+ continue;
+ }
+
+ if (seen.Add(observation.DataItemId))
+ {
+ touched.Add(observation.DataItemId);
+ }
+ }
+
+ FanOut(touched, index);
+ }
+
+ ///
+ /// Publishes the index's current snapshot of to every handle
+ /// that subscribes them. References nobody subscribed raise nothing: the Agent streams the
+ /// whole device, and republishing all of it would flood the server with values no client
+ /// asked for.
+ ///
+ private void FanOut(IReadOnlyList references, MTConnectObservationIndex index)
+ {
+ if (references.Count == 0)
+ {
+ return;
+ }
+
+ SubscriptionState[] states;
+ lock (_subscriptionLock)
+ {
+ if (_subscriptions.Count == 0)
+ {
+ return;
+ }
+
+ states = [.. _subscriptions.Values];
+ }
+
+ // The lock is released before any callback: handlers are caller code.
+ foreach (var reference in references)
+ {
+ DataValueSnapshot? snapshot = null;
+
+ foreach (var state in states)
+ {
+ if (!state.References.Contains(reference))
+ {
+ continue;
+ }
+
+ // Read once per reference, so every handle sees the identical snapshot.
+ snapshot ??= index.Get(reference);
+ RaiseDataChange(state.Handle, reference, snapshot);
+ }
+ }
+ }
+
+ ///
+ /// Raises one data-change callback, absorbing anything the subscriber throws. A consumer bug
+ /// must not tear down the pump that serves every OTHER subscriber — and the exception is
+ /// logged rather than swallowed silently, because a handler that throws on every value is
+ /// otherwise invisible.
+ ///
+ private void RaiseDataChange(MTConnectSampleHandle handle, string reference, DataValueSnapshot snapshot)
+ {
+ try
+ {
+ OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, reference, snapshot));
+ }
+ catch (Exception ex)
+ {
+ _logger.LogWarning(
+ ex,
+ "MTConnect driver {DriverInstanceId} caught a fault from a data-change subscriber for {Reference} on {SubscriptionId}; the stream continues.",
+ _driverInstanceId, reference, handle.DiagnosticId);
+ }
+ }
+
+ /// Every reference any live subscription is interested in.
+ private List SubscribedReferences()
+ {
+ lock (_subscriptionLock)
+ {
+ var all = new HashSet(StringComparer.Ordinal);
+ foreach (var state in _subscriptions.Values)
+ {
+ all.UnionWith(state.References);
+ }
+
+ return [.. all];
+ }
+ }
+
+ ///
+ /// Whether any live subscription actually names a reference. Caller must hold
+ /// . A subscription over an empty reference list is legal
+ /// (a caller that adds tags later) and must not open a stream by itself.
+ ///
+ private bool HasSubscribedReferencesCore()
+ {
+ foreach (var state in _subscriptions.Values)
+ {
+ if (state.References.Count > 0)
+ {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ ///
+ /// Records the Agent instanceId a re-baseline observed, onto whichever session is
+ /// installed. A compare-and-swap for the same reason
+ /// uses one: if a lifecycle change lands mid-update,
+ /// its state is newer and this write must be dropped, not replayed over it.
+ ///
+ private void PublishAgentInstanceId(long instanceId)
+ {
+ while (true)
+ {
+ var session = Volatile.Read(ref _session);
+ if (session is null || session.InstanceId == instanceId)
+ {
+ return;
+ }
+
+ var updated = session with { InstanceId = instanceId };
+ if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session))
+ {
+ return;
+ }
+ }
+ }
+
///
/// 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
@@ -687,17 +1414,34 @@ public sealed class MTConnectDriver : IDriver, IReadable
_options = options;
built = null;
- // One publish, so a concurrent lock-free reader sees the new client and the new probe
- // cache together or neither of them.
- Volatile.Write(ref _session, new AgentSession(client, probe, CountDataItems(probe)));
- _agentInstanceId = current.InstanceId;
+ // 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.
+ Volatile.Write(
+ ref _session,
+ new AgentSession(client, probe, CountDataItems(probe), current.InstanceId, current.NextSequence));
var index = new MTConnectObservationIndex(options.Tags);
index.Apply(current);
Volatile.Write(ref _index, index);
+ // 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;
+ _readDegraded = false;
+
WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
+ // Standing subscriptions survive a lifecycle restart; everything they hold came from the
+ // baseline just thrown away, so the restarted pump republishes as its first act. It is
+ // the PUMP that republishes rather than this method, deliberately: raising subscriber
+ // callbacks from here would run caller code on the thread holding the non-reentrant
+ // lifecycle semaphore, and a handler that touched the driver's lifecycle would deadlock.
+ lock (_subscriptionLock)
+ {
+ _streamUnsupported = false;
+ StartSampleStreamCore(republishOnStart: true);
+ }
+
_logger.LogInformation(
"MTConnect driver {DriverInstanceId} initialized against {AgentUri}{DeviceScope}: {DeviceCount} device(s), {ObservationCount} primed observation(s), agent instanceId {InstanceId}.",
_driverInstanceId,
@@ -739,8 +1483,6 @@ public sealed class MTConnectDriver : IDriver, IReadable
Volatile.Write(ref _session, null);
SafeDispose(session?.Client);
- _agentInstanceId = null;
-
// A fresh index rather than Clear(): the tag table it coerces against belongs to the options
// that are being torn down.
Volatile.Write(ref _index, new MTConnectObservationIndex(_options.Tags));
@@ -749,21 +1491,85 @@ public sealed class MTConnectDriver : IDriver, IReadable
}
///
- /// Stops the shared /sample long-poll stream. A no-op until Task 11 introduces the
- /// pump; it exists now because both and
- /// must stop the stream before the client they are about
- /// to dispose goes away, and that ordering is easy to lose when the pump is bolted on later.
+ /// Stops the shared /sample long-poll stream and waits for the pump to actually be
+ /// gone. Idempotent, and never throws — it runs on teardown paths where a second exception
+ /// would mask the first.
///
///
- /// ⚠️ This runs while is held. Whatever Task 11 fills it with
- /// — cancelling the pump's token, awaiting its task — must not call back into
- /// , or
- /// , directly or by awaiting a pump that is itself blocked on one
- /// of them: the semaphore is non-reentrant and the result is a permanent hang with no
- /// exception and no log. See the remarks for the re-baseline
- /// pattern that avoids this.
+ ///
+ /// It waits, rather than merely signalling. Every caller is about to dispose the
+ /// client the pump is enumerating (or install a new observation index the old pump would
+ /// write into). Returning while the pump is still unwinding would let it publish one more
+ /// value into state its owner has already retired — the exact "torn down but still
+ /// reporting" shape the lifecycle is written to prevent.
+ ///
+ ///
+ /// ⚠️ This runs while is held (from
+ /// , and
+ /// ) — and it awaits the pump. That is only safe because the
+ /// pump never touches : its re-baseline calls
+ /// directly (see
+ /// ) rather than re-entering a lifecycle method, and every
+ /// await it makes observes the pump token cancelled below. Break either property and
+ /// this await is a permanent hang with no exception and no log line.
+ ///
+ ///
+ /// The one thing outside that guarantee is a subscriber callback: an
+ /// handler that blocks forever blocks teardown with it. The
+ /// pump raises callbacks outside every lock precisely so that a slow handler only
+ /// delays; an infinite one is a consumer bug this driver cannot paper over.
+ ///
///
- private static Task StopSampleStreamAsync() => Task.CompletedTask;
+ private async Task StopSampleStreamAsync()
+ {
+ CancellationTokenSource? cts;
+ Task? task;
+
+ lock (_subscriptionLock)
+ {
+ cts = _pumpCts;
+ task = _pumpTask;
+ _pumpCts = null;
+ _pumpTask = null;
+ }
+
+ if (cts is null)
+ {
+ return;
+ }
+
+ try
+ {
+ await cts.CancelAsync().ConfigureAwait(false);
+ }
+ catch (ObjectDisposedException)
+ {
+ // Raced another stop; the pump is already going away.
+ }
+
+ if (task is not null)
+ {
+ try
+ {
+ await task.ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ // RunSampleStreamAsync is written not to throw, so this is belt-and-braces — but a
+ // teardown that rethrew would replace whatever failure prompted it.
+ _logger.LogDebug(
+ ex,
+ "MTConnect driver {DriverInstanceId} swallowed a fault from its /sample pump while stopping the stream.",
+ _driverInstanceId);
+ }
+ }
+
+ cts.Dispose();
+
+ // There is no stream any more, so there is nothing for it to be degraded about. Leaving the
+ // flag set would pin a later successful read to Degraded forever.
+ _streamDegraded = false;
+ }
///
/// Resolves the options a lifecycle call should apply, reporting an unreadable document
@@ -943,11 +1749,8 @@ public sealed class MTConnectDriver : IDriver, IReadable
///
private void DegradeAfterReadFailure(Exception ex)
{
- var current = ReadHealth();
- if (current.State != DriverState.Faulted)
- {
- WriteHealth(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message));
- }
+ _readDegraded = true;
+ Degrade(ex);
_logger.LogWarning(
ex,
@@ -955,6 +1758,57 @@ public sealed class MTConnectDriver : IDriver, IReadable
_driverInstanceId, _options.AgentUri);
}
+ /// Publishes without downgrading a Faulted driver.
+ private void Degrade(Exception ex)
+ {
+ var current = ReadHealth();
+ if (current.State != DriverState.Faulted)
+ {
+ WriteHealth(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message));
+ }
+ }
+
+ ///
+ /// Publishes — unless the other data path is currently
+ /// failing.
+ ///
+ ///
+ ///
+ /// Health precedence, stated deliberately. This driver has two independent Agent
+ /// data paths: /current (the batch read) and
+ /// /sample (the subscription pump). They are different HTTP requests and fail
+ /// separately — a proxy that collapses multipart/x-mixed-replace breaks streaming
+ /// while reads stay perfect; an Agent that stalls composing a whole-device snapshot
+ /// breaks reads while the stream keeps flowing.
+ ///
+ ///
+ /// The naive last-writer-wins publish would let either path paper over the other's
+ /// failure, and "Healthy" while half the surface returns Bad is precisely the
+ /// failure-wearing-success shape this codebase exists to avoid. So each path owns a
+ /// flag, clears only its own, and Healthy requires both clear: the driver reports
+ /// Degraded while anything is failing and recovers as soon as the failing path
+ /// itself recovers. — a config-level verdict — is
+ /// never downgraded by either.
+ ///
+ ///
+ /// When the Agent was last successfully heard from.
+ private void PublishHealthy(DateTime at)
+ {
+ var current = ReadHealth();
+
+ if (_streamDegraded || _readDegraded)
+ {
+ if (current.State != DriverState.Faulted)
+ {
+ WriteHealth(new DriverHealth(DriverState.Degraded, at, current.LastError));
+ }
+
+ return;
+ }
+
+ WriteHealth(new DriverHealth(DriverState.Healthy, at, null));
+ }
+
///
/// Disposes an agent client without letting a disposal fault escape. Teardown runs on the
/// failure path of , where a second exception would replace the
@@ -1050,10 +1904,53 @@ public sealed class MTConnectDriver : IDriver, IReadable
/// Data items declared by , counted once at cache time and
/// always 0 when it is null.
///
+ ///
+ /// The Agent's instanceId as last observed. Updated in place (by
+ /// ) when the pump sees the Agent restart, so the
+ /// session always names the Agent process the client is actually talking to.
+ ///
+ ///
+ /// The sequence the /sample pump should open at — the priming /current's
+ /// nextSequence. Lives here so that starting the pump reads the client and its cursor
+ /// as one consistent pair; a torn read would open the stream at another session's sequence.
+ ///
private sealed record AgentSession(
IMTConnectAgentClient Client,
MTConnectProbeModel? ProbeModel,
- int ProbeModelDataItemCount);
+ int ProbeModelDataItemCount,
+ long InstanceId,
+ long NextSequence);
+
+ ///
+ /// One live subscription: its handle and the references it wants. Both are effectively
+ /// immutable once registered, so the pump reads without a lock —
+ /// it takes only to snapshot the registry itself.
+ ///
+ /// The identity handed to the caller and stamped on every event.
+ /// The de-duplicated DataItem ids this subscription publishes.
+ private sealed record SubscriptionState(MTConnectSampleHandle Handle, HashSet References);
+
+ /// How one /sample connection ended, and therefore what the pump does next.
+ private enum StreamVerdict
+ {
+ /// Cancelled by teardown — stop, publish nothing, report nothing.
+ Cancelled = 0,
+
+ /// A transient failure: degrade and reconnect under the backoff.
+ Transient = 1,
+
+ /// Re-baselined successfully: reopen immediately at the new cursor, no backoff.
+ Reopen = 2,
+
+ /// The endpoint cannot stream at all: latch, report loudly, and stop retrying.
+ Unsupported = 3
+ }
+
+ /// How the connection ended.
+ /// The sequence the next connection must open at.
+ /// The Agent instance the cursor belongs to.
+ /// The failure, when there was one.
+ private sealed record StreamOutcome(StreamVerdict Verdict, long Cursor, long InstanceId, Exception? Error);
// ---- config DTOs ----
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectSampleHandle.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectSampleHandle.cs
new file mode 100644
index 00000000..ee0119ed
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectSampleHandle.cs
@@ -0,0 +1,29 @@
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
+
+///
+/// Driver-internal identity of one /sample subscription, matching the sibling drivers'
+/// shape (Galaxy's GalaxySubscriptionHandle, the shared PollGroupEngine's
+/// PollSubscriptionHandle): a monotonic per-driver id plus a diagnostic string that
+/// carries it into logs.
+///
+///
+///
+/// Every handle shares ONE Agent stream. Unlike a polled driver, where each
+/// subscription owns a loop, MTConnect's /sample long poll is per-Agent: the driver
+/// opens exactly one and fans each chunk out to whichever handles subscribe the reporting
+/// DataItem. The handle is therefore purely an identity — it owns no connection, no task,
+/// and no cursor — and dropping one only stops the stream when it was the last.
+///
+///
+/// A for value equality on the id, so a handle that has round-tripped
+/// through the caller still resolves. The id is never reused within a driver instance.
+///
+///
+/// The monotonic per-driver subscription id.
+internal sealed record MTConnectSampleHandle(long SubscriptionId) : ISubscriptionHandle
+{
+ ///
+ public string DiagnosticId => $"mtconnect-sub-{SubscriptionId}";
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs
index 71fe5e52..d3c2d20d 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs
@@ -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;
///
internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
{
- private readonly Channel _chunks = Channel.CreateUnbounded();
private readonly CancellationTokenSource _disposeCts = new();
+ /// Chunks a test has scripted but not yet pumped — see .
+ private readonly ConcurrentQueue _script = new();
+
+ ///
+ /// The current /sample stream generation. Replaced (not merely completed) by
+ /// 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.
+ ///
+ private Channel _chunks = Channel.CreateUnbounded();
+
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));
+ ///
+ /// An Agent scripted for the ring-buffer-overflow story: the driver primes from
+ /// Fixtures/current.xml (nextSequence 108), then the first scripted chunk is
+ /// Fixtures/sample-gap.xml — whose firstSequence (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.
+ ///
+ ///
+ /// The second /current answer is what makes this a story rather than a single
+ /// event: it advertises the buffer the gap chunk revealed (firstSequence 5000 /
+ /// nextSequence 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.
+ ///
+ 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;
+ }
+
/// The document /probe answers with. Settable so a test can re-shape the model.
public MTConnectProbeModel Probe { get; set; }
@@ -81,6 +130,31 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
/// When set, /current throws this instead of answering.
public Exception? CurrentFailure { get; set; }
+ ///
+ /// Scripted /current answers, consumed in order: each call dequeues the next one and
+ /// promotes it to , 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.
+ ///
+ public ConcurrentQueue CurrentAnswers { get; } = new();
+
+ ///
+ /// Scripted one-shot /sample failures: each enumeration dequeues one and throws it.
+ /// Models a transient stream fault the Agent recovers from — e.g. the
+ /// a real cppagent's OUT_OF_RANGE error document
+ /// (served under HTTP 200) surfaces as.
+ ///
+ public ConcurrentQueue SampleFailures { get; } = new();
+
+ ///
+ /// A sticky /sample failure: every enumeration throws it, forever. Models a
+ /// configuration-level fault such as
+ /// , where reconnecting reproduces the
+ /// identical answer — the shape a pump must NOT retry-loop against.
+ ///
+ public Exception? SampleFailure { get; set; }
+
///
/// When set, the next /probe 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
///
public TaskCompletionSource? CurrentGate { get; set; }
+ ///
+ /// Signalled the moment a /current request lands — before
+ /// 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 on a timer.
+ ///
+ public TaskCompletionSource? CurrentEntered { get; set; }
+
/// Number of /probe requests issued against this client.
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
}
///
- /// Closes the scripted /sample stream, which surfaces to the consumer as an
- /// — the transient, reconnectable end.
+ /// Queues chunks for to deliver one at a time. Nothing is sent until
+ /// a test asks for it — the script is a plan, not a schedule.
///
- public void EndStream() => _chunks.Writer.TryComplete();
+ /// The chunks the Agent should send, in order.
+ public void ScriptChunks(params MTConnectStreamsResult[] chunks)
+ {
+ ArgumentNullException.ThrowIfNull(chunks);
+
+ foreach (var chunk in chunks)
+ {
+ _script.Enqueue(chunk);
+ }
+ }
+
+ ///
+ /// Pumps the next scripted chunk (see ) and waits until the
+ /// consumer has finished with it.
+ ///
+ /// A task completing once the consumer has processed — or abandoned the stream on — that chunk.
+ /// The script is exhausted.
+ public Task PumpOnce() =>
+ _script.TryDequeue(out var next)
+ ? PumpAsync(next)
+ : throw new InvalidOperationException(
+ "No scripted /sample chunk left to pump; call ScriptChunks first.");
+
+ ///
+ /// Closes the current scripted /sample stream, which surfaces to its consumer as an
+ /// — the transient, reconnectable end — and opens
+ /// a fresh one so a reconnecting consumer has somewhere to land.
+ ///
+ public void EndStream()
+ {
+ var closed = Interlocked.Exchange(ref _chunks, Channel.CreateUnbounded());
+ closed.Writer.TryComplete();
+ }
///
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();
}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSubscribeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSubscribeTests.cs
new file mode 100644
index 00000000..d7b950b7
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSubscribeTests.cs
@@ -0,0 +1,827 @@
+using System.Collections.Concurrent;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
+
+///
+/// Task 11 — 's half: one shared
+/// /sample long-poll pump behind every handle, the ring-buffer re-baseline, and the
+/// teardown that must leave nothing running. Every test runs against
+/// ; no socket is opened anywhere in this file.
+///
+///
+///
+/// Nothing here waits on wall-clock time. The fake completes a
+/// only once the pump has finished with that
+/// chunk, and exposes the pump's own task
+/// () as the teardown barrier — so every
+/// "the pump has now done X" statement is a fact, not a sleep. The one
+/// below is a failure guard that turns a genuine hang (an
+/// un-terminated retry loop, a deadlocked teardown) into a clean red test; no assertion
+/// is ever made about elapsed time.
+///
+///
+/// The load-bearing tests are the ones about a stream that has gone wrong. A pump
+/// that only ever meets contiguous chunks is a dozen lines; the defects live in the four
+/// ways it can be knocked off the sequence — the buffer rolling past the cursor
+/// (IsSequenceGap), the Agent refusing an evicted from outright
+/// (OUT_OF_RANGE under HTTP 200 ⇒ ), the Agent
+/// restarting (a new instanceId, which ALSO looks like a gap), and the connection
+/// dropping — plus the one failure it must NOT retry
+/// (, where reconnecting reproduces the
+/// identical answer forever).
+///
+///
+public sealed class MTConnectSubscribeTests
+{
+ private const string AgentUri = "http://fixture-agent:5000";
+
+ /// The instanceId every canned fixture shares.
+ private const long FixtureInstanceId = 1655000000L;
+
+ /// A DIFFERENT instanceId — the Agent came back as a new process.
+ private const long RestartedInstanceId = 1655999999L;
+
+ /// The cursor Fixtures/current.xml primes the pump with.
+ private const long PrimedNextSequence = 108L;
+
+ // Canonical Opc.Ua.StatusCodes numerics, restated so the test asserts the wire value a client
+ // actually sees rather than a driver-private constant it could drift with.
+ private const uint Good = 0x00000000u;
+ private const uint BadNoCommunication = 0x80310000u;
+ private const uint BadWaitingForInitialData = 0x80320000u;
+ private const uint BadNodeIdUnknown = 0x80340000u;
+
+ ///
+ /// Failure guard for the handful of awaits that a defect could turn into a permanent hang
+ /// (a NotSupported retry loop, a teardown that deadlocks on the lifecycle semaphore).
+ /// Deliberately far longer than any of these operations could legitimately take — it exists
+ /// to make a hang red, not to assert a duration.
+ ///
+ private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(10);
+
+ private static readonly DateTime ObservedAt = new(2026, 7, 24, 12, 30, 0, DateTimeKind.Utc);
+
+ private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+ private static MTConnectDriverOptions Opts(MTConnectReconnectOptions? reconnect = null) =>
+ new()
+ {
+ AgentUri = AgentUri,
+ RequestTimeoutMs = 5000,
+ Reconnect = reconnect ?? new MTConnectReconnectOptions(),
+ Tags =
+ [
+ new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64),
+ new MTConnectTagDefinition("dev1_execution", DriverDataType.String),
+ new MTConnectTagDefinition("dev1_partcount", DriverDataType.Int32),
+ new MTConnectTagDefinition("dev1_never_reported", DriverDataType.Int32),
+ ],
+ };
+
+ // ---- fixtures ----
+
+ private static (MTConnectDriver Driver, CannedAgentClient Client) NewDriver(
+ CannedAgentClient? client = null, MTConnectDriverOptions? options = null)
+ {
+ var agent = client ?? CannedAgentClient.FromFixtures();
+
+ return (new MTConnectDriver(options ?? Opts(), "mt1", _ => agent), agent);
+ }
+
+ private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> InitializedDriverAsync(
+ CannedAgentClient? client = null, MTConnectDriverOptions? options = null)
+ {
+ var (driver, agent) = NewDriver(client, options);
+ await driver.InitializeAsync("{}", Ct);
+
+ return (driver, agent);
+ }
+
+ /// Attaches a recorder to the driver's data-change event and returns its log.
+ private static ConcurrentQueue Record(MTConnectDriver driver)
+ {
+ var seen = new ConcurrentQueue();
+ driver.OnDataChange += (_, e) => seen.Enqueue(e);
+
+ return seen;
+ }
+
+ /// Builds a /sample chunk (or /current answer) for the fixture Agent.
+ private static MTConnectStreamsResult Chunk(
+ long firstSequence, long nextSequence, params (string Id, string Value)[] observations) =>
+ ChunkFrom(FixtureInstanceId, firstSequence, nextSequence, observations);
+
+ private static MTConnectStreamsResult ChunkFrom(
+ long instanceId, long firstSequence, long nextSequence, params (string Id, string Value)[] observations) =>
+ new(
+ instanceId,
+ nextSequence,
+ firstSequence,
+ [.. observations.Select(o => new MTConnectObservation(o.Id, o.Value, ObservedAt))]);
+
+ private static IReadOnlyList For(
+ ConcurrentQueue seen, string reference) =>
+ [.. seen.Where(e => e.FullReference == reference)];
+
+ // ---- initial data ----
+
+ ///
+ /// The plan's first TDD case, and the OPC UA Part 4 convention: a subscription reports the
+ /// current value immediately, out of the index the priming /current already filled —
+ /// it does not make the caller wait for the Agent's next chunk, which may be a whole
+ /// heartbeat away.
+ ///
+ [Fact]
+ public async Task Subscribe_fires_initial_data_from_current()
+ {
+ var (driver, _) = await InitializedDriverAsync();
+ var seen = Record(driver);
+
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ seen.Select(e => e.FullReference).ShouldContain("dev1_pos");
+ var initial = seen.Single();
+ initial.Snapshot.StatusCode.ShouldBe(Good);
+ initial.Snapshot.Value.ShouldBe(123.4567d);
+ }
+
+ ///
+ /// Initial data fires for EVERY subscribed reference, including the ones with no value yet.
+ /// A driver that fired only for the references it happens to hold a value for would leave
+ /// the others with no snapshot at all — indistinguishable, from the server's side, from a
+ /// subscription that never established.
+ ///
+ [Fact]
+ public async Task Subscribe_fires_initial_data_for_every_ref_including_the_valueless_ones()
+ {
+ var (driver, _) = await InitializedDriverAsync();
+ var seen = Record(driver);
+
+ await driver.SubscribeAsync(
+ ["dev1_pos", "dev1_partcount", "dev1_never_reported", "nope"], TimeSpan.FromMilliseconds(50), Ct);
+
+ seen.Count.ShouldBe(4);
+ For(seen, "dev1_pos")[0].Snapshot.StatusCode.ShouldBe(Good);
+ For(seen, "dev1_partcount")[0].Snapshot.StatusCode.ShouldBe(BadNoCommunication);
+ For(seen, "dev1_never_reported")[0].Snapshot.StatusCode.ShouldBe(BadWaitingForInitialData);
+ For(seen, "nope")[0].Snapshot.StatusCode.ShouldBe(BadNodeIdUnknown);
+ }
+
+ /// Every event carries the handle the caller was given — that is how it demultiplexes.
+ [Fact]
+ public async Task Subscribe_returns_a_distinct_handle_per_call_and_stamps_it_on_every_event()
+ {
+ var (driver, _) = await InitializedDriverAsync();
+ var seen = Record(driver);
+
+ var a = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ var b = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ a.ShouldNotBe(b);
+ a.DiagnosticId.ShouldNotBe(b.DiagnosticId);
+ a.DiagnosticId.ShouldNotBeNullOrWhiteSpace();
+ For(seen, "dev1_pos").Select(e => e.SubscriptionHandle).ShouldBe([a, b]);
+ }
+
+ /// Subscribing to nothing costs the Agent nothing — no stream is opened for it.
+ [Fact]
+ public async Task Subscribe_of_an_empty_ref_list_opens_no_stream()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+
+ var handle = await driver.SubscribeAsync([], TimeSpan.FromMilliseconds(50), Ct);
+
+ handle.ShouldNotBeNull();
+ client.SampleCallCount.ShouldBe(0);
+
+ // Asserted on the pump itself, not only on the call count: a pump that HAD been started
+ // might simply not have reached its first request yet, and would pass the count alone.
+ driver.SampleStreamTask.ShouldBeNull();
+
+ // …and it still unsubscribes cleanly, so a caller that conditionally adds refs later is symmetric.
+ await driver.UnsubscribeAsync(handle, Ct);
+ }
+
+ // ---- the shared stream ----
+
+ ///
+ /// One Agent, one /sample long poll — no matter how many handles. MTConnect streams
+ /// the whole device, so a stream per subscription would multiply the Agent's connection load
+ /// by the number of OPC UA subscriptions for identical data.
+ ///
+ [Fact]
+ public async Task Two_subscriptions_share_exactly_one_sample_stream()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ await driver.SubscribeAsync(["dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
+
+ // Barrier: a chunk can only be consumed by a running enumeration, so both subscriptions have
+ // demonstrably landed on the same one by the time this returns.
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+ client.SampleCallCount.ShouldBe(1);
+ client.LastSampleFrom.ShouldBe(PrimedNextSequence);
+ }
+
+ ///
+ /// A chunk updates the index for everything it carries, but only the subscribed references
+ /// raise a callback. The Agent streams every DataItem on the device; publishing all of them
+ /// would flood the server with values nothing asked for.
+ ///
+ [Fact]
+ public async Task Pump_raises_OnDataChange_only_for_subscribed_refs()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ seen.Clear();
+
+ await client.PumpAsync(CannedAgentClient.Chunk("Fixtures/sample.xml")).WaitAsync(Watchdog, Ct);
+
+ seen.Select(e => e.FullReference).Distinct().ShouldBe(["dev1_pos"]);
+ For(seen, "dev1_pos")[0].Snapshot.Value.ShouldBe(124.01d);
+
+ // The index took the whole chunk even though only one ref was published.
+ driver.ObservationIndex.Get("dev1_partcount").Value.ShouldBe(42);
+ }
+
+ /// Each handle hears about the references IT subscribed, and only those.
+ [Fact]
+ public async Task Pump_fans_each_observation_to_every_handle_that_subscribes_it()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ var a = await driver.SubscribeAsync(["dev1_pos", "dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
+ var b = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ seen.Clear();
+
+ await client
+ .PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"), ("dev1_execution", "READY")))
+ .WaitAsync(Watchdog, Ct);
+
+ For(seen, "dev1_pos").Select(e => e.SubscriptionHandle).ShouldBe([a, b], ignoreOrder: true);
+ For(seen, "dev1_execution").Select(e => e.SubscriptionHandle).ShouldBe([a]);
+ }
+
+ ///
+ /// The running-cursor pin. An observation-free heartbeat still advances the sequence —
+ /// the Agent sends one precisely so a quiet connection can be told from a dead one — and the
+ /// gap check must be made against the PREVIOUS chunk's nextSequence, never against the
+ /// from the stream was opened with. A driver that compares against the opening
+ /// from, or that skips an empty chunk when advancing, reports a gap on the third chunk
+ /// here and re-baselines against a perfectly healthy stream.
+ ///
+ [Fact]
+ public async Task Pump_advances_the_cursor_on_every_chunk_including_an_observation_free_one()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ seen.Clear();
+
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+ // A heartbeat: no observations, and firstSequence is the buffer floor, not the cursor.
+ await client.PumpAsync(Chunk(1, 120)).WaitAsync(Watchdog, Ct);
+
+ // Contiguous with the heartbeat's nextSequence — a gap ONLY if the heartbeat was ignored.
+ await client.PumpAsync(Chunk(120, 125, ("dev1_pos", "2.5"))).WaitAsync(Watchdog, Ct);
+
+ client.CurrentCallCount.ShouldBe(1); // the initialize prime, and nothing else
+ client.SampleCallCount.ShouldBe(1); // one uninterrupted stream
+ For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(2.5d);
+ }
+
+ // ---- ring-buffer overflow ----
+
+ ///
+ /// The plan's second TDD case. The Agent's buffer rolled past the cursor
+ /// (firstSequence 5000 > the driver's 108), so the observations in between are
+ /// gone: the driver must re-baseline from /current rather than trust an incremental
+ /// update — and must do it once, then resume, not once per chunk forever.
+ ///
+ [Fact]
+ public async Task Sequence_gap_triggers_exactly_one_current_rebaseline_then_resumes()
+ {
+ var client = CannedAgentClient.WithGapThenResume();
+ var (driver, _) = await InitializedDriverAsync(client);
+ var seen = Record(driver);
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ seen.Clear();
+
+ await client.PumpOnce().WaitAsync(Watchdog, Ct); // the gap chunk
+
+ client.CurrentCallCount.ShouldBeGreaterThan(1); // the plan's assertion
+ client.CurrentCallCount.ShouldBe(2); // prime + exactly one re-baseline
+
+ await client.PumpOnce().WaitAsync(Watchdog, Ct); // the contiguous follow-up
+
+ client.CurrentCallCount.ShouldBe(2); // no re-baseline storm
+ client.SampleCallCount.ShouldBe(2); // the stream was reopened once
+ For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(201.5d);
+ }
+
+ ///
+ /// …and it resumes from the re-baselined nextSequence, not from the stale
+ /// cursor the Agent has already evicted. Reopening at the old cursor would earn the identical
+ /// rejection immediately and forever.
+ ///
+ [Fact]
+ public async Task Sequence_gap_resumes_the_stream_from_the_rebaselined_next_sequence()
+ {
+ var client = CannedAgentClient.WithGapThenResume();
+ var (driver, _) = await InitializedDriverAsync(client);
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ await client.PumpOnce().WaitAsync(Watchdog, Ct); // gap
+ await client.PumpOnce().WaitAsync(Watchdog, Ct); // resume — only a reopened stream delivers this
+
+ client.LastSampleFrom.ShouldBe(5005L);
+ }
+
+ ///
+ /// The other half of the overflow story. A real cppagent answers a from that
+ /// has already fallen out of its buffer with an MTConnectError / OUT_OF_RANGE
+ /// document served under HTTP 200 — which the client surfaces as an
+ /// , never as a gap-bearing chunk. A pump that re-baselines
+ /// only on IsSequenceGap therefore fails precisely when it has fallen furthest behind.
+ ///
+ [Fact]
+ public async Task Out_of_range_stream_error_also_triggers_a_rebaseline()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+
+ client.SampleFailures.Enqueue(new InvalidDataException(
+ "MTConnect /sample answered an MTConnectError document (OUT_OF_RANGE) under HTTP 200"));
+ client.CurrentAnswers.Enqueue(Chunk(490, 500, ("dev1_pos", "50.5")));
+
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ seen.Clear();
+
+ // Only a pump that re-baselined and reopened the stream can consume this.
+ await client.PumpAsync(Chunk(500, 505, ("dev1_pos", "77.5"))).WaitAsync(Watchdog, Ct);
+
+ client.CurrentCallCount.ShouldBe(2);
+ client.SampleCallCount.ShouldBe(2);
+ client.LastSampleFrom.ShouldBe(500L);
+ For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(77.5d);
+ }
+
+ // ---- agent restart ----
+
+ ///
+ /// The instanceId check must come BEFORE the gap check. An Agent restart changes
+ /// instanceId AND resets sequences, so a restart usually trips
+ /// IsSequenceGap too — and the two demand different handling: a gap keeps the held
+ /// values (they are still this device's), a restart invalidates every one of them because
+ /// the device model they describe no longer exists. This chunk is deliberately BOTH: new
+ /// instanceId and a firstSequence far past the cursor. A driver that tests the gap
+ /// first keeps serving dev1_execution's pre-restart value indefinitely — the new
+ /// Agent never reports it again, so nothing would ever overwrite it.
+ ///
+ [Fact]
+ public async Task Agent_restart_clears_the_index_and_is_detected_before_the_gap_path()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ await driver.SubscribeAsync(["dev1_pos", "dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
+ seen.Clear();
+
+ // The restarted Agent's /current knows nothing about dev1_execution.
+ client.CurrentAnswers.Enqueue(ChunkFrom(RestartedInstanceId, 40, 42, ("dev1_pos", "9.5")));
+
+ await client
+ .PumpAsync(ChunkFrom(RestartedInstanceId, 5000, 5005, ("dev1_pos", "5.5")))
+ .WaitAsync(Watchdog, Ct);
+
+ driver.AgentInstanceId.ShouldBe(RestartedInstanceId);
+ client.CurrentCallCount.ShouldBe(2);
+
+ // Cleared: the pre-restart value is gone, not merely shadowed.
+ driver.ObservationIndex.Get("dev1_execution").StatusCode.ShouldBe(BadWaitingForInitialData);
+ driver.ObservationIndex.Get("dev1_pos").Value.ShouldBe(9.5d);
+
+ // And the subscriber was TOLD its value went away, rather than being left holding a stale Good.
+ For(seen, "dev1_execution").Last().Snapshot.StatusCode.ShouldBe(BadWaitingForInitialData);
+ For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(9.5d);
+ }
+
+ // ---- stream ends ----
+
+ ///
+ /// A dropped connection is transient: reconnect from the cursor and keep going. It is
+ /// emphatically NOT a re-baseline — the sequence is still valid, and spending a
+ /// /current on every network blip would be a self-inflicted load.
+ ///
+ [Fact]
+ public async Task Stream_ended_reconnects_from_the_cursor_without_a_rebaseline()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ seen.Clear();
+
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+ client.EndStream();
+
+ // Only a reconnected pump can consume a chunk written to the NEW stream generation.
+ await client.PumpAsync(Chunk(113, 118, ("dev1_pos", "2.5"))).WaitAsync(Watchdog, Ct);
+
+ client.SampleCallCount.ShouldBe(2);
+ client.LastSampleFrom.ShouldBe(113L);
+ client.CurrentCallCount.ShouldBe(1);
+ For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(2.5d);
+ }
+
+ ///
+ /// The hot-loop pin. means the
+ /// endpoint will never stream to this request — a reverse proxy, a health page, an Agent
+ /// fronted by something that collapses multipart/x-mixed-replace. Reconnecting
+ /// reproduces the identical answer forever, so the pump must give up loudly instead of
+ /// hammering the endpoint until an operator notices. A later subscription must not restart
+ /// the loop either: the answer has not changed just because someone asked again.
+ ///
+ [Fact]
+ public async Task Stream_not_supported_stops_the_pump_and_never_retries()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ client.SampleFailure = new MTConnectStreamNotSupportedException(
+ "the Agent answered /sample with a single non-multipart document");
+
+ var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ var pump = driver.SampleStreamTask.ShouldNotBeNull();
+
+ await pump.WaitAsync(Watchdog, Ct);
+
+ client.SampleCallCount.ShouldBe(1);
+ driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+ driver.GetHealth().LastError.ShouldNotBeNullOrWhiteSpace();
+
+ // A full unsubscribe/resubscribe cycle must not re-open the wound either.
+ await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ client.SampleCallCount.ShouldBe(1);
+ }
+
+ ///
+ /// …but a re-initialize IS an operator changing something, so it clears the verdict and
+ /// tries again. Otherwise fixing the proxy would need a process restart.
+ ///
+ [Fact]
+ public async Task Reinitialize_clears_the_stream_unsupported_verdict()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ client.SampleFailure = new MTConnectStreamNotSupportedException("not multipart");
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ await driver.SampleStreamTask!.WaitAsync(Watchdog, Ct);
+
+ client.SampleFailure = null;
+ await driver.ReinitializeAsync("{}", Ct).WaitAsync(Watchdog, Ct);
+
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+ client.SampleCallCount.ShouldBe(2);
+ }
+
+ // ---- backoff ----
+
+ ///
+ /// The reconnect backoff as a pure function of the attempt number: the first retry honours
+ /// MinBackoffMs (zero = immediate), and every later one grows geometrically to the
+ /// cap. The 100 ms growth floor matters more than it looks — the default
+ /// MinBackoffMs is 0, and 0 × multiplier is still 0, so without a floor
+ /// the "geometric" backoff would be an unbounded hot loop against a dead Agent.
+ ///
+ [Fact]
+ public void Reconnect_backoff_starts_at_min_grows_geometrically_and_caps()
+ {
+ var options = new MTConnectReconnectOptions
+ {
+ MinBackoffMs = 0, MaxBackoffMs = 1000, BackoffMultiplier = 2.0,
+ };
+
+ MTConnectDriver.BackoffFor(1, options).ShouldBe(TimeSpan.Zero);
+ MTConnectDriver.BackoffFor(2, options).ShouldBe(TimeSpan.FromMilliseconds(200));
+ MTConnectDriver.BackoffFor(3, options).ShouldBe(TimeSpan.FromMilliseconds(400));
+ MTConnectDriver.BackoffFor(4, options).ShouldBe(TimeSpan.FromMilliseconds(800));
+ MTConnectDriver.BackoffFor(5, options).ShouldBe(TimeSpan.FromMilliseconds(1000));
+ MTConnectDriver.BackoffFor(500, options).ShouldBe(TimeSpan.FromMilliseconds(1000));
+ }
+
+ ///
+ /// Operator-authored nonsense must not un-bound the loop: a multiplier of 1 (or less) would
+ /// never grow, and a negative delay is not a delay. Both fall back to the shipped default.
+ ///
+ [Fact]
+ public void Reconnect_backoff_survives_a_degenerate_multiplier()
+ {
+ var options = new MTConnectReconnectOptions
+ {
+ MinBackoffMs = -5, MaxBackoffMs = 10_000, BackoffMultiplier = 0.5,
+ };
+
+ MTConnectDriver.BackoffFor(1, options).ShouldBe(TimeSpan.Zero);
+ MTConnectDriver.BackoffFor(2, options).ShouldBeGreaterThan(TimeSpan.Zero);
+ MTConnectDriver.BackoffFor(3, options)
+ .ShouldBeGreaterThan(MTConnectDriver.BackoffFor(2, options));
+ }
+
+ // ---- unsubscribe ----
+
+ ///
+ /// Dropping one handle drops only its references; the shared stream keeps running for the
+ /// others. Tearing the stream down on the first unsubscribe would silently blind every
+ /// remaining subscription.
+ ///
+ [Fact]
+ public async Task Unsubscribe_drops_only_that_handles_refs_and_keeps_the_shared_stream()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ var a = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ var b = await driver.SubscribeAsync(["dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
+ var pump = driver.SampleStreamTask.ShouldNotBeNull();
+
+ await driver.UnsubscribeAsync(a, Ct).WaitAsync(Watchdog, Ct);
+ seen.Clear();
+
+ await client
+ .PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"), ("dev1_execution", "READY")))
+ .WaitAsync(Watchdog, Ct);
+
+ pump.IsCompleted.ShouldBeFalse();
+ seen.Select(e => e.FullReference).Distinct().ShouldBe(["dev1_execution"]);
+
+ await driver.UnsubscribeAsync(b, Ct).WaitAsync(Watchdog, Ct);
+
+ pump.IsCompleted.ShouldBeTrue();
+ driver.SampleStreamTask.ShouldBeNull();
+ }
+
+ /// The last unsubscribe stops the stream — and nothing is published after it.
+ [Fact]
+ public async Task Unsubscribe_of_the_last_handle_stops_the_stream_and_silences_the_driver()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
+ seen.Clear();
+
+ // The stream is gone, so this chunk has no consumer and nothing may reach the recorder.
+ client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).IsCompleted.ShouldBeFalse();
+
+ seen.ShouldBeEmpty();
+ driver.SampleStreamTask.ShouldBeNull();
+ }
+
+ ///
+ /// Unsubscribing twice, or with a handle this driver never issued, is a no-op — not a throw.
+ /// Teardown paths run on failure paths, and an exception there would mask the real fault.
+ ///
+ [Fact]
+ public async Task Unsubscribe_is_idempotent_and_ignores_a_foreign_handle()
+ {
+ var (driver, _) = await InitializedDriverAsync();
+ var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
+ await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
+ await driver.UnsubscribeAsync(new ForeignHandle(), Ct).WaitAsync(Watchdog, Ct);
+ }
+
+ /// A null argument is a caller bug, and the one thing these methods are loud about.
+ [Fact]
+ public async Task Null_arguments_throw_ArgumentNullException()
+ {
+ var (driver, _) = await InitializedDriverAsync();
+
+ await Should.ThrowAsync(
+ () => driver.SubscribeAsync(null!, TimeSpan.FromMilliseconds(50), Ct));
+ await Should.ThrowAsync(() => driver.UnsubscribeAsync(null!, Ct));
+ }
+
+ // ---- lifecycle ----
+
+ ///
+ /// Subscribing before the driver is initialized registers the interest and reports the
+ /// honest "no value yet" — it does not throw, and it does not dial an Agent that does not
+ /// exist yet. Initialize then picks the standing subscription up.
+ ///
+ [Fact]
+ public async Task Subscribe_before_initialize_registers_and_initialize_starts_the_stream()
+ {
+ var (driver, client) = NewDriver();
+ var seen = Record(driver);
+
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ client.SampleCallCount.ShouldBe(0);
+ driver.SampleStreamTask.ShouldBeNull();
+ seen.Single().Snapshot.StatusCode.ShouldBe(BadWaitingForInitialData);
+
+ await driver.InitializeAsync("{}", Ct);
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+ client.SampleCallCount.ShouldBe(1);
+ For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(1.5d);
+ }
+
+ ///
+ /// A re-initialize replaces the served state, so it must replace the pump with it: the old
+ /// one is stopped (it holds the old cursor, and possibly the old client) and a new one starts
+ /// from the fresh /current. The standing subscription survives — the caller did not
+ /// ask for it to be dropped — and is re-primed, because every value it holds came from a
+ /// baseline that has just been thrown away.
+ ///
+ [Fact]
+ public async Task Reinitialize_stops_the_old_pump_starts_a_new_one_and_republishes()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ var first = driver.SampleStreamTask.ShouldNotBeNull();
+
+ // Barrier: the first pump is demonstrably enumerating before the re-initialize lands on it.
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+ client.CurrentAnswers.Enqueue(Chunk(300, 310, ("dev1_pos", "310.5")));
+ seen.Clear();
+
+ await driver.ReinitializeAsync("{}", Ct).WaitAsync(Watchdog, Ct);
+
+ first.IsCompleted.ShouldBeTrue();
+ var second = driver.SampleStreamTask.ShouldNotBeNull();
+ second.ShouldNotBeSameAs(first);
+
+ await client.PumpAsync(Chunk(310, 315, ("dev1_pos", "315.5"))).WaitAsync(Watchdog, Ct);
+
+ client.SampleCallCount.ShouldBe(2);
+ client.LastSampleFrom.ShouldBe(310L);
+ For(seen, "dev1_pos").Select(e => e.Snapshot.Value).ShouldBe([310.5d, 315.5d]);
+ }
+
+ ///
+ /// Shutdown stops the pump BEFORE disposing the client. Getting that order wrong leaves the
+ /// pump enumerating a disposed client — which it would read as a dropped connection and try
+ /// to reconnect, forever, against an object that no longer exists.
+ ///
+ [Fact]
+ public async Task Shutdown_stops_the_pump_before_disposing_the_client_and_leaks_nothing()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ var pump = driver.SampleStreamTask.ShouldNotBeNull();
+
+ // Barrier: the stream is demonstrably open before the shutdown lands, so the call count
+ // below is a statement about reconnect churn rather than about who won a race.
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+ seen.Clear();
+
+ await driver.ShutdownAsync(Ct).WaitAsync(Watchdog, Ct);
+
+ pump.IsCompleted.ShouldBeTrue();
+ pump.IsFaulted.ShouldBeFalse();
+ driver.SampleStreamTask.ShouldBeNull();
+ client.IsDisposed.ShouldBeTrue();
+
+ // No reconnect churn against the disposed client, and nothing published after teardown.
+ client.SampleCallCount.ShouldBe(1);
+ seen.ShouldBeEmpty();
+ }
+
+ ///
+ /// The deadlock pin. The lifecycle semaphore is non-reentrant, so a pump that reached
+ /// back into InitializeAsync/ReinitializeAsync/ShutdownAsync — the
+ /// natural way to write "on a gap, just re-prime" — would hang the driver permanently, with
+ /// no exception and no log. Here the pump is parked INSIDE its re-baseline /current
+ /// when a shutdown lands and takes that semaphore: the shutdown must cancel the pump and
+ /// complete. If the pump were waiting on the lifecycle lock instead, neither side would ever
+ /// move.
+ ///
+ [Fact]
+ public async Task Shutdown_while_the_pump_is_mid_rebaseline_does_not_deadlock()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ client.CurrentEntered = entered;
+ client.CurrentGate = release;
+
+ // A gap forces the re-baseline; the gate holds the pump inside the /current request.
+ _ = client.PumpAsync(Chunk(5000, 5005, ("dev1_pos", "5.5")));
+ await entered.Task.WaitAsync(Watchdog, Ct);
+
+ try
+ {
+ await driver.ShutdownAsync(Ct).WaitAsync(Watchdog, Ct);
+ }
+ finally
+ {
+ release.TrySetResult();
+ }
+
+ driver.SampleStreamTask.ShouldBeNull();
+ client.IsDisposed.ShouldBeTrue();
+ }
+
+ // ---- robustness + health ----
+
+ ///
+ /// A subscriber that throws is a bug in the consumer, not a reason to stop delivering data to
+ /// every other subscriber. The pump absorbs it and keeps pumping.
+ ///
+ [Fact]
+ public async Task A_throwing_subscriber_does_not_kill_the_pump()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ var seen = Record(driver);
+ driver.OnDataChange += (_, _) => throw new InvalidOperationException("subscriber blew up");
+
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ seen.Clear();
+
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+ await client.PumpAsync(Chunk(113, 118, ("dev1_pos", "2.5"))).WaitAsync(Watchdog, Ct);
+
+ driver.SampleStreamTask!.IsCompleted.ShouldBeFalse();
+ For(seen, "dev1_pos").Select(e => e.Snapshot.Value).ShouldBe([1.5d, 2.5d]);
+ }
+
+ ///
+ /// Health precedence, read side. A healthy stream must not paper over a broken read
+ /// path: /sample and /current are different Agent requests and can fail
+ /// independently, and "Healthy" while every OPC UA Read returns Bad is exactly the
+ /// failure-wearing-success shape this driver is written against.
+ ///
+ [Fact]
+ public async Task A_pumped_chunk_does_not_paper_over_a_failed_read()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+
+ client.CurrentFailure = new HttpRequestException("connection refused");
+ await driver.ReadAsync(["dev1_pos"], Ct);
+ driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+ driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+
+ // …and clearing the read failure does restore it, because the stream never failed.
+ client.CurrentFailure = null;
+ await driver.ReadAsync(["dev1_pos"], Ct);
+ driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+ }
+
+ ///
+ /// Health precedence, stream side. The mirror image: a successful read must not paper
+ /// over a stream that has stopped delivering. Only the path that failed may clear its own
+ /// degradation.
+ ///
+ [Fact]
+ public async Task A_successful_read_does_not_paper_over_a_broken_stream()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ client.SampleFailure = new MTConnectStreamNotSupportedException("not multipart");
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ await driver.SampleStreamTask!.WaitAsync(Watchdog, Ct);
+
+ var res = await driver.ReadAsync(["dev1_pos"], Ct);
+
+ res[0].StatusCode.ShouldBe(Good);
+ driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+ }
+
+ /// A recovered stream clears its own degradation once chunks flow again.
+ [Fact]
+ public async Task A_recovered_stream_restores_health()
+ {
+ var (driver, client) = await InitializedDriverAsync();
+ client.SampleFailures.Enqueue(new TimeoutException("no chunk within the heartbeat window"));
+
+ await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
+ await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
+
+ driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+ client.SampleCallCount.ShouldBe(2);
+ }
+
+ /// A handle from some other driver — the type check must not be a cast.
+ private sealed class ForeignHandle : ISubscriptionHandle
+ {
+ public string DiagnosticId => "not-ours";
+ }
+}