From dfbcb0e7f0519426a38499965d80825fa90146d1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 24 Jul 2026 17:16:13 -0400 Subject: [PATCH] fix(mtconnect): close the silent-refusal + teardown-wedge defects in the pump (Task 11 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit C1 (critical): a latched "this endpoint cannot stream" verdict could end with a GREEN driver holding a permanently silent subscription. DriverInstanceActor handles a tag-set change as Unsubscribe-then-Subscribe with no re-initialize; the unsubscribe cleared the stream-degraded flag, the resubscribe was refused by the latch without a word, and one successful read then reported Healthy. StartSampleStreamCore now re-asserts the degradation and logs a Warning naming the remedy, and teardown no longer clears the flag while the latch is set. NOT promoted to Faulted: DriverHealthReport maps any Faulted driver to a /readyz 503, so that would de-ready the whole node over one misconfigured Agent whose reads are perfectly healthy. Faulted stays for the case that earns it. I1: teardown waits are bounded (5 s) and honour the caller's token. They were unbounded and ignored it, so DriverInstanceActor's 5 s budget was inert and PostStop — which blocks a dispatcher thread on ShutdownAsync while holding the lifecycle semaphore — could be wedged forever by one blocking subscriber. The cancellation source is disposed only when the loop is provably gone. Applied to StopProbeLoopAsync too: Task 13 reproduced the identical shape, and closing the class in one method while leaving the other as the pattern to copy is worse than not closing it. I2: the reconnect ladder was monotonic for the process lifetime, so ~9 unrelated drops pinned the driver at MaxBackoffMs forever. A stream that delivered before dropping now starts a fresh ladder (at attempt 1, not 0, so MinBackoffMs is still honoured). I3: MaxBackoffMs <= 0 clamped every delay to zero — an operator-authorable reconnect spin. Treated as unset, like a multiplier that cannot grow. I4: the session published instanceId without NextSequence, so after a restart it held the new agent's id beside the dead one's cursor, and a gap/OUT_OF_RANGE re-baseline (id unchanged) never published the cursor at all. Both move in one CAS, and the pump publishes its cursor as it exits. I5: OnDataChange was a multicast Invoke — the first throwing handler aborted the rest of the list, starving every later subscriber. Now walked by hand with the catch inside the loop. The test that claimed to cover this registered the recorder BEFORE the thrower, so it passed regardless; swapped, it was red. Faults are latched to one Warning per stream generation (Debug thereafter) so a consistently-throwing consumer cannot flood the log. Also: the pump restarts after an unexpected fault (IsCompleted, not null); a device-scope that matches nothing is a Warning, not a Debug tally; a device or component with neither name nor id is skipped rather than emitting a blank path segment; DiscoverAsync's remark no longer claims DriverInstanceActor retries (it does not for a Once driver, and injection is dormant in v3); and two flake seeds in the fake are gone (Dispose re-read its own counter; _disposeCts was disposed under a racing SampleAsync). 439/439. Every changed behaviour falsified by mutation. --- .../MTConnectDriver.cs | 442 +++++++++++++++--- .../CannedAgentClient.cs | 69 ++- .../MTConnectDiscoverTests.cs | 104 ++++- .../MTConnectHostAndRediscoverTests.cs | 47 ++ .../MTConnectSubscribeTests.cs | 277 ++++++++++- .../RecordingDriverLogger.cs | 65 +++ 6 files changed, 926 insertions(+), 78 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/RecordingDriverLogger.cs 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]; + } + } +}