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 7a129af8..6b6c4f77 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs @@ -103,6 +103,31 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// Growth factor used when the authored multiplier could not grow the delay (≤ 1). private const double DefaultBackoffMultiplier = 2.0; + /// + /// Backoff cap used when is authored + /// non-positive. Treated as "unset" rather than as "no cap" for the same reason a + /// multiplier ≤ 1 falls back to : a zero cap clamps + /// EVERY delay to zero, which turns the reconnect loop into a spin against a refused + /// connection — one Warning per iteration, as fast as the socket can fail. Mirrors the + /// shipped default. + /// + private const int DefaultMaxBackoffMs = 30_000; + + /// + /// How long teardown waits for a background loop (the /sample pump, the connectivity + /// probe) to unwind before abandoning the wait and carrying on. + /// + /// + /// An unbounded wait here is not merely slow — it wedges an actor-system thread. + /// DriverInstanceActor.PostStop calls with + /// GetAwaiter().GetResult(), i.e. a synchronous block on an Akka dispatcher thread, + /// while this driver holds . So a subscriber callback that never + /// returns would take the dispatcher thread AND every subsequent lifecycle call on this + /// driver with it, permanently. Five seconds matches the budget + /// DriverInstanceActor already puts around . + /// + private static readonly TimeSpan TeardownWait = TimeSpan.FromSeconds(5); + /// /// Config-JSON reader options, mirroring the sibling driver factories. Note there is /// deliberately no JsonStringEnumConverter: enum-carrying DTO fields stay @@ -265,6 +290,21 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// Whether the /current read path is currently failing. See . private volatile bool _readDegraded; + /// + /// Consecutive /sample reconnect attempts that have not yet been vindicated by + /// a delivered chunk — the input to . Reset by any stream that + /// actually delivered something, so an agent that drops a connection once an hour is never + /// treated as one that has been failing all day. + /// + private volatile int _reconnectAttempts; + + /// + /// Latch behind the once-per-stream-generation subscriber-fault Warning. A consumer that + /// throws on every value would otherwise emit one Warning per observation per handle + /// — thousands a second on a busy Agent, which buries the very log it is trying to raise. + /// + private int _subscriberFaultLogged; + /// /// The live agent client plus the /probe cache derived from it, held as one /// immutable snapshot behind a single reference and only ever replaced by a @@ -376,6 +416,21 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// internal long? AgentInstanceId => Volatile.Read(ref _session)?.InstanceId; + /// + /// The sequence the next /sample stream will open at — the priming /current's + /// nextSequence, advanced by every re-baseline and by the pump as it stops. Paired + /// with in one session record, so the two are never observed + /// out of step. + /// + internal long? AgentNextSequence => Volatile.Read(ref _session)?.NextSequence; + + /// + /// Consecutive /sample reconnect attempts not yet vindicated by a delivered chunk — + /// the input to , exposed so the reset policy can be asserted + /// without measuring how long anything slept. + /// + internal int ReconnectAttempts => _reconnectAttempts; + /// /// 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, @@ -428,8 +483,8 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis // 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 StopProbeLoopAsync().ConfigureAwait(false); + await StopSampleStreamAsync(cancellationToken).ConfigureAwait(false); + await StopProbeLoopAsync(cancellationToken).ConfigureAwait(false); await TeardownCoreAsync().ConfigureAwait(false); await StartCoreAsync(options, existingClient: null, cancellationToken).ConfigureAwait(false); @@ -473,8 +528,8 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis var existing = Volatile.Read(ref _session)?.Client; - await StopSampleStreamAsync().ConfigureAwait(false); - await StopProbeLoopAsync().ConfigureAwait(false); + await StopSampleStreamAsync(cancellationToken).ConfigureAwait(false); + await StopProbeLoopAsync(cancellationToken).ConfigureAwait(false); if (existing is null || RequiresClientRebuild(_options, incoming)) { @@ -506,8 +561,8 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis { var lastRead = ReadHealth().LastSuccessfulRead; - await StopSampleStreamAsync().ConfigureAwait(false); - await StopProbeLoopAsync().ConfigureAwait(false); + await StopSampleStreamAsync(cancellationToken).ConfigureAwait(false); + await StopProbeLoopAsync(cancellationToken).ConfigureAwait(false); await TeardownCoreAsync().ConfigureAwait(false); WriteHealth(new DriverHealth(DriverState.Unknown, lastRead, null)); @@ -797,7 +852,9 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis { // 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); + // The caller's token bounds the wait: DriverInstanceActor puts a 5 s budget around this + // call, and until now that budget was inert. + await StopSampleStreamAsync(cancellationToken).ConfigureAwait(false); } _logger.LogDebug( @@ -886,9 +943,22 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// discovery stream has no such channel, so the only way to "fail quietly" would be to /// emit an empty tree — indistinguishable from an Agent that genuinely declares nothing, /// and read by an operator in the picker as a working connection to an empty machine - /// (#485: empty is not an answer). Both production callers handle the throw: the - /// universal browser surfaces it as a failed browse, and DriverInstanceActor logs - /// it and retries on its rediscover tick. + /// (#485: empty is not an answer). + /// + /// + /// What the callers actually do with the throw — worth stating precisely, because + /// the reassuring version ("it is logged and retried") is not true here. The live + /// consumer is the /raw browse-commit path + /// (DiscoveryDriverBrowser/BrowserSessionService), which surfaces it to the + /// operator as a failed browse — that is the case this posture is chosen for. + /// DriverInstanceActor.HandleRediscoverAsync catches it, logs a Warning, and + /// publishes an EMPTY node set; it does not reschedule, because retrying is a + /// behaviour and this driver is + /// . And that publish currently reaches + /// nothing: DriverHostActor.HandleDiscoveredNodes short-circuits unconditionally + /// (discovered-node injection is dormant in v3 — raw tags are authored through + /// browse-commit instead), so at that seam the throw-versus-empty choice has no runtime + /// effect today. It is made for the browse path, and for whatever re-enables injection. /// /// /// Nothing here writes the observation index. Discovery is a pure read of the @@ -920,11 +990,37 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis continue; } - devices++; var name = FolderName(device.Name, device.Id); + if (string.IsNullOrWhiteSpace(name)) + { + // Neither a name nor an id: there is no browse path that could identify this device, + // and folding it in would emit a blank tree segment (and a blank NodeId component + // downstream). Its data items are unreachable either way. + _logger.LogWarning( + "MTConnect driver {DriverInstanceId} skipped a device in {AgentUri}'s /probe model that declares neither a name nor an id; it cannot be given a browse path.", + _driverInstanceId, _options.AgentUri); + + continue; + } + + devices++; variables += StreamContainer(device, builder.Folder(name, name)); } + // A configured scope that matches NO device is an authoring error — almost always a typo or + // a device renamed in the Agent — and its only symptom is a browse that succeeds and shows + // nothing. Reported at Warning, because at Debug the operator's evidence is an empty picker + // and no explanation. An agent-wide browse that finds nothing is a different statement (the + // Agent declares no devices) and stays a Debug tally. + if (devices == 0 && !string.IsNullOrEmpty(deviceScope)) + { + _logger.LogWarning( + "MTConnect driver {DriverInstanceId} is scoped to device '{DeviceScope}', but {AgentUri}'s /probe model declares no device with that name or id ({DeclaredCount} declared). The browse is empty because the scope matched nothing, not because the Agent has nothing.", + _driverInstanceId, deviceScope, _options.AgentUri, model.Devices.Count); + + return; + } + _logger.LogDebug( "MTConnect driver {DriverInstanceId} streamed {DeviceCount} device(s) and {VariableCount} data item(s) from {AgentUri}'s /probe model{DeviceScope}.", _driverInstanceId, @@ -997,7 +1093,18 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis continue; } + // Guarded on the RESOLVED folder name, not on the id alone: a component with a blank id + // but a real name browses perfectly well (the id is only the fallback), while one with + // neither would open a folder named "" — a blank segment in every browse path beneath + // it. The data-item guard above is stricter for a different reason: THAT id is the + // correlation key an observation is resolved by, so a blank one is unreadable whatever + // it is called. var name = FolderName(component.Name, component.Id); + if (string.IsNullOrWhiteSpace(name)) + { + continue; + } + streamed += StreamContainer(component, scope.Folder(name, name)); } @@ -1244,10 +1351,12 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// ⚠️ This runs while is held and it awaits the loop, on /// exactly the same terms as : safe only because the loop /// never touches the lifecycle and every await it makes observes the token cancelled below. - /// The one thing outside that guarantee is an handler that - /// blocks forever — a consumer bug this driver cannot paper over. + /// The wait is bounded by and the caller's token for the same + /// reason: an handler that never returns would otherwise + /// wedge the Akka dispatcher thread DriverInstanceActor.PostStop blocks on. /// - private async Task StopProbeLoopAsync() + /// The caller's deadline for the teardown wait. + private async Task StopProbeLoopAsync(CancellationToken cancellationToken = default) { var cts = _probeCts; var task = Volatile.Read(ref _probeTask); @@ -1268,11 +1377,21 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis // Raced another stop; the loop is already going away. } + var stopped = true; if (task is not null) { try { - await task.ConfigureAwait(false); + await task.WaitAsync(TeardownWait, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is TimeoutException or OperationCanceledException) + { + stopped = false; + + _logger.LogError( + ex, + "MTConnect driver {DriverInstanceId} gave up waiting for its connectivity probe to stop after {TeardownWait}; the usual cause is an OnHostStatusChanged subscriber that blocks. Teardown continues.", + _driverInstanceId, TeardownWait); } catch (Exception ex) { @@ -1284,7 +1403,11 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis } } - cts.Dispose(); + if (stopped) + { + // Only once the loop is provably gone — see StopSampleStreamAsync for why. + cts.Dispose(); + } } // ---- IRediscoverable: the Agent instanceId watch ---- @@ -1395,7 +1518,33 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// private void StartSampleStreamCore(bool republishOnStart) { - if (_pumpTask is not null || _streamUnsupported || !HasSubscribedReferencesCore()) + // `IsCompleted: false`, not `is not null`: a pump that has already exited is not a running + // stream, and conflating the two means a pump killed by an unexpected fault could never be + // revived by a resubscribe. The genuinely-unrepeatable case is the latch below, not this. + if (_pumpTask is { IsCompleted: false }) + { + return; + } + + if (_streamUnsupported) + { + // Refusing IS correct — reconnecting reproduces the identical answer forever — but it + // must never be silent. DriverInstanceActor handles a tag-set change as + // Unsubscribe-then-Subscribe with NO re-initialize, and the unsubscribe stops a stream + // that is not running; without this, the resubscribe returns a handle, the caller is + // told SubscriptionEstablished, and the next successful read reports a perfectly green + // driver over a subscription that can never deliver a value (#485). + _streamDegraded = true; + Degrade($"The MTConnect Agent at '{_options.AgentUri}' does not serve a framed /sample stream; subscriptions cannot deliver until the driver is re-initialized."); + + _logger.LogWarning( + "MTConnect driver {DriverInstanceId} refused to start a /sample stream for {AgentUri}: the endpoint was already found not to stream. The subscription is registered but will receive NO updates until the driver is re-initialized.", + _driverInstanceId, _options.AgentUri); + + return; + } + + if (!HasSubscribedReferencesCore()) { return; } @@ -1408,6 +1557,15 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis return; } + // A previous pump ran to completion; its cancellation source is ours to release before the + // field is overwritten, or every revived stream leaks one. + _pumpCts?.Dispose(); + _pumpTask = null; + + // A new stream generation gets a fresh subscriber-fault Warning budget (see + // _subscriberFaultLogged). + Interlocked.Exchange(ref _subscriberFaultLogged, 0); + var options = _options; var cts = new CancellationTokenSource(); _pumpCts = cts; @@ -1450,6 +1608,7 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis { var cursor = from; var attempt = 0; + _reconnectAttempts = 0; try { @@ -1507,18 +1666,26 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis case StreamVerdict.Reopen: attempt = 0; + _reconnectAttempts = 0; break; default: _streamDegraded = true; Degrade(outcome.Error!); - attempt++; + + // A stream that DELIVERED before it dropped proves the endpoint works, so + // its failure starts a fresh backoff ladder. Without this the counter is + // monotonic for the life of the process: a healthy agent that drops a + // connection once an hour would be pinned at MaxBackoffMs within a day, and + // "the first retry is immediate" would be true exactly once ever. + attempt = outcome.DeliveredAny ? 1 : attempt + 1; + _reconnectAttempts = 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); + "MTConnect driver {DriverInstanceId} lost the /sample stream from {AgentUri}; reconnecting from sequence {Cursor} (attempt {Attempt}, delivered={Delivered}).", + _driverInstanceId, options.AgentUri, cursor, attempt, outcome.DeliveredAny); break; } @@ -1534,6 +1701,16 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis _streamDegraded = true; Degrade(ex); } + finally + { + // The session's cursor is the opening `from` of the NEXT pump, and until now it only + // ever held the priming /current's. A last-unsubscribe + resubscribe (no re-initialize) + // would therefore reopen at a sequence the Agent has long since evicted: it self-heals + // via OUT_OF_RANGE, but only after a wasted round trip, and a still-buffered stale + // cursor replays historical observations to subscribers first. One publish per pump + // lifetime, dropped if a lifecycle change has already installed a newer session. + PublishAgentCursor(instanceId, cursor); + } } /// @@ -1554,6 +1731,11 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis // endless /current re-baseline storm against a perfectly healthy stream. var expected = cursor; + // Whether THIS connection ever handed us a chunk. A stream that delivered and then dropped + // is a working endpoint having a bad moment; one that never delivered may be an endpoint + // that cannot work at all. They must not share a backoff ladder — see RunSampleStreamAsync. + var delivered = false; + try { await foreach (var chunk in client.SampleAsync(cursor, ct).ConfigureAwait(false)) @@ -1586,7 +1768,8 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis // DiscoverAsync. AnnounceAgentRestart(chunk.InstanceId, options); - return await RebaselineAsync(client, options, expected, instanceId, ct).ConfigureAwait(false); + return await RebaselineAsync(client, options, expected, instanceId, delivered, ct) + .ConfigureAwait(false); } if (IMTConnectAgentClient.IsSequenceGap(expected, chunk)) @@ -1595,10 +1778,12 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis "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); + return await RebaselineAsync(client, options, expected, instanceId, delivered, ct) + .ConfigureAwait(false); } ApplyAndFanOut(chunk); + delivered = true; // 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 @@ -1614,21 +1799,22 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis // 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.Cancelled, expected, instanceId, delivered, null) : new StreamOutcome( StreamVerdict.Transient, expected, instanceId, + delivered, new MTConnectStreamEndedException( MTConnectStreamEndReason.ConnectionClosed, options.AgentUri, 0)); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { - return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, null); + return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null); } catch (MTConnectStreamNotSupportedException ex) { - return new StreamOutcome(StreamVerdict.Unsupported, expected, instanceId, ex); + return new StreamOutcome(StreamVerdict.Unsupported, expected, instanceId, delivered, ex); } catch (InvalidDataException ex) { @@ -1642,13 +1828,14 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis "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); + return await RebaselineAsync(client, options, expected, instanceId, delivered, 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); + return new StreamOutcome(StreamVerdict.Transient, expected, instanceId, delivered, ex); } } @@ -1670,6 +1857,7 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis MTConnectDriverOptions options, long fallbackCursor, long fallbackInstanceId, + bool deliveredAny, CancellationToken ct) { try @@ -1677,7 +1865,7 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, ct) .ConfigureAwait(false); - PublishAgentInstanceId(current.InstanceId); + PublishAgentCursor(current.InstanceId, current.NextSequence); ApplyAndFanOut(current); _streamDegraded = false; @@ -1691,11 +1879,13 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis current.InstanceId, current.NextSequence); - return new StreamOutcome(StreamVerdict.Reopen, current.NextSequence, current.InstanceId, null); + return new StreamOutcome( + StreamVerdict.Reopen, current.NextSequence, current.InstanceId, deliveredAny, null); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { - return new StreamOutcome(StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, null); + return new StreamOutcome( + StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, deliveredAny, null); } catch (Exception ex) { @@ -1703,7 +1893,8 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis // 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); + return new StreamOutcome( + StreamVerdict.Transient, fallbackCursor, fallbackInstanceId, deliveredAny, ex); } } @@ -1717,7 +1908,9 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// 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. + /// , a non-positive cap falls back to + /// (clamping to it would make EVERY delay zero — a spin, + /// not a backoff), and a negative minimum clamps to zero. /// /// The 1-based reconnect attempt about to be made. /// The authored backoff options. @@ -1726,7 +1919,9 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis { ArgumentNullException.ThrowIfNull(reconnect); - var capMs = Math.Max(0, reconnect.MaxBackoffMs); + // A non-positive cap is "unset", NOT "no cap". Math.Max(0, ...) would clamp every delay to + // zero and turn the reconnect loop into a spin against a refused connection. + var capMs = reconnect.MaxBackoffMs > 0 ? reconnect.MaxBackoffMs : DefaultMaxBackoffMs; var minMs = Math.Max(0, reconnect.MinBackoffMs); if (attempt <= 1) @@ -1829,24 +2024,73 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis } /// - /// 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. + /// Raises one data-change callback to every subscriber, absorbing anything each one + /// throws. A consumer bug must not tear down the pump that serves the others — nor rob them + /// of the value. /// + /// + /// + /// The invocation list is walked by hand, with the try/catch INSIDE the loop. A + /// plain OnDataChange?.Invoke(...) is a multicast call: the first handler that + /// throws aborts the rest of the list, so one broken consumer silently starves every + /// other subscriber of that observation — and a single try/catch around the whole + /// invoke absorbs the exception while leaving the starvation in place, which reads as + /// "isolated" in a log and is not. + /// + /// + /// One is shared across the list deliberately: it is + /// an immutable record, and allocating per handler on the hot path would buy nothing. + /// + /// private void RaiseDataChange(MTConnectSampleHandle handle, string reference, DataValueSnapshot snapshot) { - try + var handlers = OnDataChange; + if (handlers is null) { - OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, reference, snapshot)); + return; } - catch (Exception ex) + + var args = new DataChangeEventArgs(handle, reference, snapshot); + + foreach (var registration in handlers.GetInvocationList()) + { + try + { + ((EventHandler)registration).Invoke(this, args); + } + catch (Exception ex) + { + LogSubscriberFault(ex, reference, handle); + } + } + } + + /// + /// Reports a faulting data-change subscriber once per stream generation at Warning, + /// and at Debug thereafter. + /// + /// + /// A consumer that throws on every value would otherwise emit one Warning per observation + /// per handle — on a busy Agent, thousands a second, which buries the incident it is + /// reporting along with everything else in the log. Nothing is lost: the later faults are + /// still emitted, at Debug, and the latch is cleared whenever the pump (re)starts. + /// + private void LogSubscriberFault(Exception ex, string reference, MTConnectSampleHandle handle) + { + if (Interlocked.Exchange(ref _subscriberFaultLogged, 1) == 0) { _logger.LogWarning( ex, - "MTConnect driver {DriverInstanceId} caught a fault from a data-change subscriber for {Reference} on {SubscriptionId}; the stream continues.", + "MTConnect driver {DriverInstanceId} caught a fault from a data-change subscriber for {Reference} on {SubscriptionId}; the stream continues and every other subscriber still received the value. Further subscriber faults are logged at Debug until the stream restarts.", _driverInstanceId, reference, handle.DiagnosticId); + + return; } + + _logger.LogDebug( + ex, + "MTConnect driver {DriverInstanceId} caught a further fault from a data-change subscriber for {Reference} on {SubscriptionId}.", + _driverInstanceId, reference, handle.DiagnosticId); } /// Every reference any live subscription is interested in. @@ -1883,22 +2127,44 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis } /// - /// 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. + /// Records where the Agent now is — its instanceId and the sequence a stream + /// should resume from — onto whichever session is installed. /// - private void PublishAgentInstanceId(long instanceId) + /// + /// + /// Both fields move together, in one compare-and-swap. They are one fact about + /// one Agent process, and the session's own contract is that a reader gets the client + /// and its cursor as a consistent pair. Publishing the id alone (as this did) left the + /// session holding a NEW instanceId beside the cursor of the process that had just + /// died — the exact tear the packing exists to prevent. + /// + /// + /// There is no "the id did not change, so skip" early return, because the two + /// re-baselines that do NOT change the id — a sequence gap and an OUT_OF_RANGE + /// rejection — are precisely the ones whose whole purpose is to move the cursor. + /// Suppressing the write there would leave the next pump reopening at the stale + /// sequence that had just been rejected. + /// + /// + /// Compare-and-swap for the same reason uses one: + /// a lifecycle change landing mid-update has newer state, so this write is dropped + /// rather than replayed over it. + /// + /// + /// The Agent instanceId last observed. + /// The sequence a /sample stream should resume from. + private void PublishAgentCursor(long instanceId, long nextSequence) { while (true) { var session = Volatile.Read(ref _session); - if (session is null || session.InstanceId == instanceId) + if (session is null || + (session.InstanceId == instanceId && session.NextSequence == nextSequence)) { return; } - var updated = session with { InstanceId = instanceId }; + var updated = session with { InstanceId = instanceId, NextSequence = nextSequence }; if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session)) { return; @@ -2190,13 +2456,19 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// 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. + /// The wait is bounded (, and the caller's token) — the + /// one thing outside the guarantee above is a subscriber callback, and an + /// handler that never returns would otherwise block teardown + /// forever. That is not merely slow: DriverInstanceActor.PostStop blocks an Akka + /// dispatcher thread on while this driver holds + /// , so an unbounded wait wedges an actor-system thread and + /// every later lifecycle call with it. Abandoning the wait is safe because the pump's + /// registry slots are cleared BEFORE it: a pump left running cannot be resurrected into + /// , and it is already cancelled. /// /// - private async Task StopSampleStreamAsync() + /// The caller's deadline for the teardown wait. + private async Task StopSampleStreamAsync(CancellationToken cancellationToken = default) { CancellationTokenSource? cts; Task? task; @@ -2223,11 +2495,21 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis // Raced another stop; the pump is already going away. } + var stopped = true; if (task is not null) { try { - await task.ConfigureAwait(false); + await task.WaitAsync(TeardownWait, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is TimeoutException or OperationCanceledException) + { + stopped = false; + + _logger.LogError( + ex, + "MTConnect driver {DriverInstanceId} gave up waiting for its /sample pump to stop after {TeardownWait}. The pump is cancelled and can publish nothing further, but something inside it is not returning — the usual cause is an OnDataChange subscriber that blocks. Teardown continues.", + _driverInstanceId, TeardownWait); } catch (Exception ex) { @@ -2240,11 +2522,25 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis } } - cts.Dispose(); + if (stopped) + { + // Disposed ONLY once the pump is provably gone: an abandoned pump still holds this + // source's token, and disposing underneath it turns its next linked-token creation into + // an ObjectDisposedException. One leaked CTS beats a mystery exception. + 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; + // Nothing to be degraded about once the stream is gone — UNLESS the endpoint is latched as + // unable to stream at all, in which case there is no prospect of one coming back, and + // clearing this would let the next successful read report a green driver over a subscription + // that can never deliver (#485). See StartSampleStreamCore. + lock (_subscriptionLock) + { + if (!_streamUnsupported) + { + _streamDegraded = false; + } + } } /// @@ -2435,12 +2731,28 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis } /// Publishes without downgrading a Faulted driver. - private void Degrade(Exception ex) + private void Degrade(Exception ex) => Degrade(ex.Message); + + /// + /// Publishes with an explanatory message, without + /// downgrading a Faulted driver. + /// + /// + /// Deliberately not , even for the "an operator must fix + /// this" cases such as an endpoint that cannot stream. In this codebase Faulted is not + /// merely a stronger word: DriverHealthReport maps any Faulted driver to a + /// /readyz 503, taking the whole server node out of readiness. A driver whose + /// reads are perfectly healthy and whose subscriptions are dead is degraded, not unready — + /// de-readying the node over one misconfigured Agent would be a far larger blast radius than + /// the fault it reports. Faulted stays reserved for the case that earns it: an initialize + /// that failed, after which the driver serves nothing at all. + /// + private void Degrade(string message) { var current = ReadHealth(); if (current.State != DriverState.Faulted) { - WriteHealth(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message)); + WriteHealth(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, message)); } } @@ -2598,7 +2910,7 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// /// /// The Agent's instanceId as last observed. Updated in place (by - /// ) when the pump sees the Agent restart, so the + /// ) when the pump sees the Agent restart, so the /// session always names the Agent process the client is actually talking to. /// /// @@ -2641,8 +2953,14 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// How the connection ended. /// The sequence the next connection must open at. /// The Agent instance the cursor belongs to. + /// + /// Whether this connection yielded at least one chunk before it ended. Distinguishes a + /// working endpoint having a bad moment from one that has never worked, which is what keeps + /// the reconnect backoff from ratcheting monotonically over a process lifetime. + /// /// The failure, when there was one. - private sealed record StreamOutcome(StreamVerdict Verdict, long Cursor, long InstanceId, Exception? Error); + private sealed record StreamOutcome( + StreamVerdict Verdict, long Cursor, long InstanceId, bool DeliveredAny, Exception? Error); // ---- config DTOs ---- 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 d3c2d20d..47617451 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 @@ -37,6 +37,9 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable { private readonly CancellationTokenSource _disposeCts = new(); + private readonly Lock _sampleWaitersLock = new(); + private readonly List<(int Threshold, TaskCompletionSource Waiter)> _sampleWaiters = []; + /// Chunks a test has scripted but not yet pumped — see . private readonly ConcurrentQueue _script = new(); @@ -201,6 +204,47 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable /// The from sequence the most recent /sample enumeration was opened at. public long? LastSampleFrom { get; private set; } + /// + /// A task completing once has reached + /// — the deterministic "the pump has now opened its Nth stream" + /// barrier, for the reconnect paths where no chunk is ever delivered and + /// therefore cannot be the barrier. + /// + /// The enumeration count to wait for. + public Task WaitForSampleCallsAsync(int count) + { + lock (_sampleWaitersLock) + { + if (Volatile.Read(ref _sampleCallCount) >= count) + { + return Task.CompletedTask; + } + + var waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _sampleWaiters.Add((count, waiter)); + + return waiter.Task; + } + } + + /// Releases any waiter the new call count satisfies. + private void ReleaseSampleWaiters(int reached) + { + lock (_sampleWaitersLock) + { + for (var i = _sampleWaiters.Count - 1; i >= 0; i--) + { + if (_sampleWaiters[i].Threshold > reached) + { + continue; + } + + _sampleWaiters[i].Waiter.TrySetResult(); + _sampleWaiters.RemoveAt(i); + } + } + } + /// public async Task ProbeAsync(CancellationToken ct) { @@ -258,7 +302,7 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable long from, [EnumeratorCancellation] CancellationToken ct) { ObjectDisposedException.ThrowIf(IsDisposed, this); - Interlocked.Increment(ref _sampleCallCount); + ReleaseSampleWaiters(Interlocked.Increment(ref _sampleCallCount)); LastSampleFrom = from; // Thrown before the first chunk, exactly like a real client that fails its /sample request: @@ -372,18 +416,33 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable } /// + /// + /// + /// The first-disposer test uses the value + /// returned, not a re-read of the counter. Re-reading is a race: two concurrent + /// disposes can both increment and then both read 2, so BOTH take the early return and + /// the stream is never closed — a subscription test would then hang waiting for a + /// teardown that silently did nothing. + /// + /// + /// is cancelled but deliberately NOT disposed. A + /// /sample enumeration that is starting concurrently reaches + /// CreateLinkedTokenSource(ct, _disposeCts.Token), and reading .Token on a + /// disposed source throws — a flake that would + /// surface as a random unrelated failure in whichever test happened to lose the race. + /// Leaking one cancelled source per fake is free; a cancelled source needs no disposal + /// to release anything a test cares about. + /// + /// public void Dispose() { - Interlocked.Increment(ref _disposeCount); - - if (DisposeCount > 1) + if (Interlocked.Increment(ref _disposeCount) > 1) { return; } _disposeCts.Cancel(); Volatile.Read(ref _chunks).Writer.TryComplete(); - _disposeCts.Dispose(); } private sealed record ScriptedChunk(MTConnectStreamsResult Result, TaskCompletionSource Consumed); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectDiscoverTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectDiscoverTests.cs index 369e366b..6e02e552 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectDiscoverTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectDiscoverTests.cs @@ -58,7 +58,9 @@ public sealed class MTConnectDiscoverTests }; private static (MTConnectDriver Driver, CannedAgentClient Client) NewDriver( - MTConnectDriverOptions? options = null, MTConnectProbeModel? probe = null) + MTConnectDriverOptions? options = null, + MTConnectProbeModel? probe = null, + RecordingDriverLogger? logger = null) { var client = CannedAgentClient.FromFixtures(); if (probe is not null) @@ -66,13 +68,15 @@ public sealed class MTConnectDiscoverTests client.Probe = probe; } - return (new MTConnectDriver(options ?? Opts(), "mt1", _ => client), client); + return (new MTConnectDriver(options ?? Opts(), "mt1", _ => client, logger), client); } private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> InitializedDriverAsync( - MTConnectDriverOptions? options = null, MTConnectProbeModel? probe = null) + MTConnectDriverOptions? options = null, + MTConnectProbeModel? probe = null, + RecordingDriverLogger? logger = null) { - var (driver, client) = NewDriver(options, probe); + var (driver, client) = NewDriver(options, probe, logger); await driver.InitializeAsync("{}", TestContext.Current.CancellationToken); return (driver, client); @@ -476,6 +480,93 @@ public sealed class MTConnectDiscoverTests cap.Variables.Select(v => v.Attr.FullName).ShouldBe(["dev9_a"]); } + /// + /// A configured DeviceName that matches nothing is an authoring error whose only + /// symptom is an empty picker. It must be reported at Warning — at Debug the operator + /// sees a browse that "worked" and a machine that apparently declares nothing, with no + /// evidence pointing at their typo. + /// + [Fact] + public async Task Discover_warns_when_the_configured_device_scope_matches_nothing() + { + var logger = new RecordingDriverLogger(); + var (driver, _) = await InitializedDriverAsync(Opts(deviceName: "no-such-device"), logger: logger); + + var cap = await DiscoverAsync(driver); + + cap.Variables.ShouldBeEmpty(); + logger.WarningsSnapshot() + .ShouldContain(w => w.Contains("no-such-device", StringComparison.Ordinal)); + } + + /// + /// …and an agent-wide browse that finds nothing is NOT that warning: it is a different + /// statement (the Agent declares no devices) and must not cry scope-typo. + /// + [Fact] + public async Task Discover_does_not_warn_about_scope_when_no_scope_is_configured() + { + var logger = new RecordingDriverLogger(); + var (driver, _) = await InitializedDriverAsync(probe: new MTConnectProbeModel([]), logger: logger); + + var cap = await DiscoverAsync(driver); + + cap.Variables.ShouldBeEmpty(); + logger.WarningsSnapshot().ShouldBeEmpty(); + } + + /// + /// A component declaring neither a name nor an id has no browse path at all, + /// and folding it in would open a folder called "" — a blank segment in the path of + /// everything beneath it. It is skipped; its siblings are unaffected. (A component with a + /// blank id but a real name is fine — the id is only the fallback.) + /// + [Fact] + public async Task Discover_skips_a_component_with_neither_a_name_nor_an_id() + { + var item = new MTConnectDataItem("dev9_x", null, "SAMPLE", "POSITION", null, null, null, null); + var probe = new MTConnectProbeModel( + [ + new MTConnectDevice( + "dev9", + "VMC", + [ + new MTConnectComponent(" ", " ", [], [item]), + new MTConnectComponent(" ", "Axes", [], [item]), + new MTConnectComponent("dev9_ctrl", null, [], [item]), + ], + []), + ]); + + var (driver, _) = await InitializedDriverAsync(probe: probe); + + var cap = await DiscoverAsync(driver); + + cap.Folders.Select(f => f.Path).ShouldBe(["VMC", "VMC/Axes", "VMC/dev9_ctrl"], ignoreOrder: true); + cap.Folders.ShouldAllBe(f => !f.Path.Contains("//", StringComparison.Ordinal)); + } + + /// The same rule one level up: a device with no browse path is skipped, and said so. + [Fact] + public async Task Discover_skips_a_device_with_neither_a_name_nor_an_id() + { + var item = new MTConnectDataItem("dev9_x", null, "SAMPLE", "POSITION", null, null, null, null); + var logger = new RecordingDriverLogger(); + var probe = new MTConnectProbeModel( + [ + new MTConnectDevice(" ", " ", [], [item]), + new MTConnectDevice("dev9", "VMC", [], [item]), + ]); + + var (driver, _) = await InitializedDriverAsync(probe: probe, logger: logger); + + var cap = await DiscoverAsync(driver); + + cap.Folders.Select(f => f.Path).ShouldBe(["VMC"]); + logger.WarningsSnapshot() + .ShouldContain(w => w.Contains("neither a name nor an id", StringComparison.Ordinal)); + } + // ---- the probe-cache contract (anti-wedge) ---- /// @@ -556,7 +647,10 @@ public sealed class MTConnectDiscoverTests /// no data items", which is the #485 shape (failure wearing success's clothes) and would read /// to an operator in the browse picker as a working connection to an empty machine. Both /// production callers handle the throw: the universal browser surfaces it as a failed browse, - /// and DriverInstanceActor logs it and retries. + /// and DriverInstanceActor logs it, publishes an empty node set, and does NOT retry + /// (retrying is an UntilStable behaviour; this driver is Once) — which today + /// reaches nothing anyway, since DriverHostActor's discovered-node injection is + /// dormant in v3. The browse path is the consumer this posture is chosen for. /// [Fact] public async Task Discover_before_initialize_throws_and_streams_nothing() diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectHostAndRediscoverTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectHostAndRediscoverTests.cs index 14be33e9..0b9e3e1b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectHostAndRediscoverTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectHostAndRediscoverTests.cs @@ -121,6 +121,53 @@ public sealed class MTConnectHostAndRediscoverTests await ticker.ParkedAsync().WaitAsync(Watchdog, Ct); } + /// + /// Teardown symmetry with the /sample pump (remediation I1). The probe loop is + /// the second long-lived background task in this driver, and it is stopped the same way and + /// from the same place: while the lifecycle semaphore is held, on a path + /// DriverInstanceActor.PostStop blocks an Akka dispatcher thread on. So its wait is + /// bounded and honours the caller's token too — a blocking + /// handler must not be able to wedge + /// shutdown, any more than a blocking data-change handler can. + /// + /// + /// Its blast radius is genuinely smaller than the pump's (each iteration is bounded by the + /// validated-positive Probe.Timeout, where the pump's is an unbounded long poll), but + /// leaving the two divergent would make the un-bounded one the pattern a future reader + /// copies — the defect class would be closed in one place and re-opened in the other, in the + /// same file. + /// + [Fact] + public async Task Shutdown_honours_the_callers_deadline_when_a_host_status_subscriber_blocks() + { + var (driver, _, ticker) = await ProbingDriverAsync(); + var blocking = new ManualResetEventSlim(false); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + ((IHostConnectivityProbe)driver).OnHostStatusChanged += (_, _) => + { + entered.TrySetResult(); + blocking.Wait(); + }; + + try + { + // The first successful probe transitions Unknown -> Running and raises the event, which + // now blocks inside the loop. + ticker.Tick(); + await entered.Task.WaitAsync(Watchdog, Ct); + + using var deadline = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)); + await driver.ShutdownAsync(deadline.Token).WaitAsync(Watchdog, Ct); + + driver.ProbeLoopTask.ShouldBeNull(); + } + finally + { + blocking.Set(); + } + } + private static ConcurrentQueue RecordRediscovery(MTConnectDriver driver) { var seen = new ConcurrentQueue(); 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 index d7b950b7..b12c32ff 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSubscribeTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSubscribeTests.cs @@ -84,17 +84,21 @@ public sealed class MTConnectSubscribeTests // ---- fixtures ---- private static (MTConnectDriver Driver, CannedAgentClient Client) NewDriver( - CannedAgentClient? client = null, MTConnectDriverOptions? options = null) + CannedAgentClient? client = null, + MTConnectDriverOptions? options = null, + RecordingDriverLogger? logger = null) { var agent = client ?? CannedAgentClient.FromFixtures(); - return (new MTConnectDriver(options ?? Opts(), "mt1", _ => agent), agent); + return (new MTConnectDriver(options ?? Opts(), "mt1", _ => agent, logger), agent); } private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> InitializedDriverAsync( - CannedAgentClient? client = null, MTConnectDriverOptions? options = null) + CannedAgentClient? client = null, + MTConnectDriverOptions? options = null, + RecordingDriverLogger? logger = null) { - var (driver, agent) = NewDriver(client, options); + var (driver, agent) = NewDriver(client, options, logger); await driver.InitializeAsync("{}", Ct); return (driver, agent); @@ -464,11 +468,24 @@ public sealed class MTConnectSubscribeTests driver.GetHealth().State.ShouldBe(DriverState.Degraded); driver.GetHealth().LastError.ShouldNotBeNullOrWhiteSpace(); - // A full unsubscribe/resubscribe cycle must not re-open the wound either. + // A full unsubscribe/resubscribe cycle must not re-open the wound either. This is exactly + // what DriverInstanceActor does for a tag-set change: UnsubscribeAsync then SubscribeAsync, + // with NO re-initialize in between. await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct); await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); client.SampleCallCount.ShouldBe(1); + + // ...and the driver must not read back GREEN afterwards. The unsubscribe stopped a stream + // that was never running, the resubscribe was silently refused by the latch, and one + // successful read was then enough to clear the last degradation flag — leaving a Healthy + // driver holding a subscription that can never deliver a value. That is #485 in the + // subscription plane: an established handle, an Established reply, and permanent silence. + var res = await driver.ReadAsync(["dev1_pos"], Ct); + + res[0].StatusCode.ShouldBe(Good); + driver.GetHealth().State.ShouldBe(DriverState.Degraded); + driver.GetHealth().LastError.ShouldNotBeNullOrWhiteSpace(); } /// @@ -747,8 +764,12 @@ public sealed class MTConnectSubscribeTests public async Task A_throwing_subscriber_does_not_kill_the_pump() { var (driver, client) = await InitializedDriverAsync(); - var seen = Record(driver); + // Registration ORDER is the whole point: the thrower goes FIRST. A multicast invoke aborts + // the rest of the invocation list at the first exception, so with the recorder registered + // first it would run before the throw and its assertions would hold however badly the + // driver isolates subscribers — the test would pass without testing anything. driver.OnDataChange += (_, _) => throw new InvalidOperationException("subscriber blew up"); + var seen = Record(driver); await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); seen.Clear(); @@ -819,6 +840,250 @@ public sealed class MTConnectSubscribeTests client.SampleCallCount.ShouldBe(2); } + /// + /// C1, the other half. Refusing to start the stream is correct, but it must not be + /// silent: the operator's only other evidence is a subscription that never delivers. The + /// refusal re-asserts the degradation AND says so, naming the remedy. + /// + [Fact] + public async Task A_refused_stream_start_re_asserts_degraded_and_says_so() + { + var logger = new RecordingDriverLogger(); + var (driver, client) = await InitializedDriverAsync(logger: logger); + client.SampleFailure = new MTConnectStreamNotSupportedException("not multipart"); + + var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + await driver.SampleStreamTask!.WaitAsync(Watchdog, Ct); + + await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + driver.GetHealth().State.ShouldBe(DriverState.Degraded); + logger.WarningsSnapshot() + .ShouldContain(w => w.Contains("refused to start a /sample stream", StringComparison.Ordinal)); + } + + /// + /// C1, the window the refusal path cannot cover. Dropping the last subscription stops + /// a stream, and teardown clears the stream degradation with it — but NOT while the endpoint + /// is latched as unable to stream at all. That latch is a standing fact about the Agent, + /// discovered at runtime and true whether or not anyone is subscribed right now; letting a + /// successful read clear it here would make the driver flicker green between subscriptions + /// and report the impairment only while someone happens to be listening. + /// + [Fact] + public async Task A_latched_endpoint_stays_degraded_even_with_no_subscriptions_left() + { + var (driver, client) = await InitializedDriverAsync(); + client.SampleFailure = new MTConnectStreamNotSupportedException("not multipart"); + + var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + await driver.SampleStreamTask!.WaitAsync(Watchdog, Ct); + + // No subscriptions left at all — so nothing re-asserts the degradation on a refused start. + await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct); + + var res = await driver.ReadAsync(["dev1_pos"], Ct); + + res[0].StatusCode.ShouldBe(Good); + driver.GetHealth().State.ShouldBe(DriverState.Degraded); + } + + /// + /// I1 — the caller's teardown deadline must be real. DriverInstanceActor wraps + /// this call in a 5 s and blocks an Akka dispatcher + /// thread on the shutdown path, so a wait that ignored the token would let one blocking + /// OnDataChange handler wedge an actor-system thread — while holding the lifecycle + /// semaphore, taking every later lifecycle call on this driver with it. + /// + [Fact] + public async Task Unsubscribe_honours_the_callers_deadline_when_a_subscriber_blocks() + { + var (driver, client) = await InitializedDriverAsync(); + var blocking = new ManualResetEventSlim(false); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + // Blocks only on the PUMPED value — the initial-data callback fires on the subscribe + // thread, and blocking there would prove nothing about teardown. + driver.OnDataChange += (_, e) => + { + if (e.Snapshot.Value is not 1.5d) + { + return; + } + + entered.TrySetResult(); + blocking.Wait(); + }; + + var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + try + { + _ = client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))); + await entered.Task.WaitAsync(Watchdog, Ct); + + // The pump is now stuck inside caller code. The unsubscribe must still return. + using var deadline = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)); + await driver.UnsubscribeAsync(handle, deadline.Token).WaitAsync(Watchdog, Ct); + + driver.SampleStreamTask.ShouldBeNull(); + } + finally + { + blocking.Set(); + } + } + + /// + /// I2 — the backoff ladder is per-outage, not per-process. A stream that delivered + /// before it dropped proves the endpoint works, so its failure starts a fresh ladder. Without + /// this the counter only ever climbs: an agent that drops one connection an hour would be + /// pinned at MaxBackoffMs within a day, having never once failed twice in a row. + /// + [Fact] + public async Task Reconnect_attempts_reset_when_the_stream_delivered_before_dropping() + { + var (driver, client) = await InitializedDriverAsync(); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct); + client.EndStream(); + await client.PumpAsync(Chunk(113, 118, ("dev1_pos", "2.5"))).WaitAsync(Watchdog, Ct); + client.EndStream(); + await client.PumpAsync(Chunk(118, 123, ("dev1_pos", "3.5"))).WaitAsync(Watchdog, Ct); + + // Two unrelated drops, each preceded by a delivered chunk: each starts a fresh ladder, so + // the count is 1 (this outage's first retry, which still honours MinBackoffMs) — never 2. + driver.ReconnectAttempts.ShouldBe(1); + client.SampleCallCount.ShouldBe(3); + } + + /// + /// …and the counter still accumulates when nothing is delivered, which is the case the + /// backoff exists for. Asserted as monotonic growth rather than an exact value, because the + /// pump may have advanced again between the barrier and the read. + /// + [Fact] + public async Task Reconnect_attempts_accumulate_while_the_stream_delivers_nothing() + { + // A 1 ms cap keeps the failing loop from actually sleeping; nothing here asserts a duration. + var options = Opts(new MTConnectReconnectOptions { MinBackoffMs = 0, MaxBackoffMs = 1 }); + var (driver, client) = await InitializedDriverAsync(options: options); + client.SampleFailure = new TimeoutException("no chunk within the heartbeat window"); + + var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + try + { + await client.WaitForSampleCallsAsync(3).WaitAsync(Watchdog, Ct); + var early = driver.ReconnectAttempts; + + await client.WaitForSampleCallsAsync(6).WaitAsync(Watchdog, Ct); + + early.ShouldBeGreaterThanOrEqualTo(2); + driver.ReconnectAttempts.ShouldBeGreaterThan(early); + } + finally + { + await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct); + } + } + + /// + /// I3 — a non-positive cap is an operator-authorable hot loop. Clamping to it (the + /// obvious Math.Max(0, …)) makes EVERY delay zero, so the pump reconnect-spins as fast + /// as a refused connection returns, one Warning per iteration. Treated as "unset" instead, + /// exactly like a multiplier that cannot grow. + /// + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Reconnect_backoff_treats_a_non_positive_cap_as_unset(int maxBackoffMs) + { + var options = new MTConnectReconnectOptions + { + MinBackoffMs = 0, MaxBackoffMs = maxBackoffMs, BackoffMultiplier = 2.0, + }; + + // The first retry is still immediate — that is MinBackoffMs, and it is honoured. + MTConnectDriver.BackoffFor(1, options).ShouldBe(TimeSpan.Zero); + + // Every later one must actually back off. + MTConnectDriver.BackoffFor(2, options).ShouldBeGreaterThan(TimeSpan.Zero); + MTConnectDriver.BackoffFor(9, options).ShouldBeGreaterThan(MTConnectDriver.BackoffFor(2, options)); + } + + /// + /// I4 — the session's instanceId and cursor are one fact and must move together. + /// Publishing the id alone left the session holding the NEW agent's id beside the OLD + /// agent's cursor, which is precisely the tear the record exists to prevent — and on a gap + /// or OUT_OF_RANGE re-baseline (where the id does not change at all) the cursor was + /// not published even once. + /// + [Fact] + public async Task Rebaseline_publishes_the_cursor_beside_the_instance_id() + { + var client = CannedAgentClient.WithGapThenResume(); + var (driver, _) = await InitializedDriverAsync(client); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + driver.AgentNextSequence.ShouldBe(PrimedNextSequence); + + await client.PumpOnce().WaitAsync(Watchdog, Ct); // the gap chunk + + driver.AgentNextSequence.ShouldBe(5005L); + driver.AgentInstanceId.ShouldBe(FixtureInstanceId); + } + + /// + /// …and the consequence that makes it matter: the NEXT pump opens at the re-baselined + /// cursor. A last-unsubscribe followed by a resubscribe is an ordinary tag-set change + /// (DriverInstanceActor does exactly that, with no re-initialize), and a stale cursor + /// there means reopening at a sequence the Agent has already evicted. + /// + [Fact] + public async Task A_restarted_pump_resumes_from_the_rebaselined_cursor() + { + var client = CannedAgentClient.WithGapThenResume(); + var (driver, _) = await InitializedDriverAsync(client); + var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + await client.PumpOnce().WaitAsync(Watchdog, Ct); // gap -> re-baseline to 5005 + + await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + await client.PumpOnce().WaitAsync(Watchdog, Ct); // the contiguous follow-up + + client.LastSampleFrom.ShouldBe(5005L); + } + + /// + /// I5 — one broken subscriber must not starve the others. A multicast + /// Invoke aborts the invocation list at the first exception, so a single try/catch + /// around it absorbs the exception while silently robbing every later subscriber of the + /// value — which reads as "isolated" in the log and is not. The thrower sits BETWEEN the two + /// recorders on purpose. + /// + [Fact] + public async Task Every_subscriber_receives_a_value_even_when_one_in_the_middle_throws() + { + var (driver, client) = await InitializedDriverAsync(); + var first = Record(driver); + driver.OnDataChange += (_, _) => throw new InvalidOperationException("subscriber blew up"); + var last = Record(driver); + + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + first.Clear(); + last.Clear(); + + await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct); + + For(first, "dev1_pos").Select(e => e.Snapshot.Value).ShouldBe([1.5d]); + For(last, "dev1_pos").Select(e => e.Snapshot.Value).ShouldBe([1.5d]); + } + /// A handle from some other driver — the type check must not be a cast. private sealed class ForeignHandle : ISubscriptionHandle { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/RecordingDriverLogger.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/RecordingDriverLogger.cs new file mode 100644 index 00000000..a2193020 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/RecordingDriverLogger.cs @@ -0,0 +1,65 @@ +using Microsoft.Extensions.Logging; + +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests; + +/// +/// Captures what the driver logs, so that "and it says so" can be asserted rather than assumed. +/// +/// +/// Used where the log line is the behaviour: a browse scope that matches nothing, and a +/// subscription the driver silently refuses to serve. Both are cases whose only other symptom is +/// an operator staring at an empty panel, so a fix that emits nothing is not a fix. +/// +internal sealed class RecordingDriverLogger : ILogger +{ + /// Formatted lines, in order. + public List Warnings { get; } = []; + + /// Formatted lines, in order. + public List Errors { get; } = []; + + /// + public IDisposable? BeginScope(TState state) + where TState : notnull => null; + + /// + public bool IsEnabled(LogLevel logLevel) => true; + + /// + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + ArgumentNullException.ThrowIfNull(formatter); + + var sink = logLevel switch + { + LogLevel.Warning => Warnings, + LogLevel.Error => Errors, + _ => null, + }; + + // Locked: the /sample pump logs from its own task while the test thread reads. + if (sink is null) + { + return; + } + + lock (sink) + { + sink.Add(formatter(state, exception)); + } + } + + /// A snapshot copy of , safe to enumerate while the pump runs. + public IReadOnlyList WarningsSnapshot() + { + lock (Warnings) + { + return [.. Warnings]; + } + } +}