diff --git a/CLAUDE.md b/CLAUDE.md index e52ff5e9..2f18b090 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -323,8 +323,8 @@ non-`Healthy` state **and** its `LastSuccessfulRead` has been frozen for 90 s. T reading. - **Fast path for DEFINITIVE losses (#676):** a driver implementing the opt-in `IDriverConnectionLossSource` raises `ConnectionLost` (once per connection) when its protocol proves the connection can never recover in place — OpcUaClient: - keep-alive `BadSessionIdInvalid`/`BadSessionClosed`/`BadSessionNotActivated`/`BadServerHalted` or a dropped - subscription — and `DriverInstanceActor` (attach PreStart, detach PostStop) turns it into the same disconnect only + keep-alive `BadSessionIdInvalid`/`BadSessionClosed`/`BadSessionNotActivated`/`BadServerHalted` or a subscription the + SERVER reported dropped (its publishing-stall heuristic does not raise) — and `DriverInstanceActor` (attach PreStart, detach PostStop) turns it into the same disconnect only while Connected; ambiguous failures (timeouts, `BadNoCommunication`, a stalled socket) must never raise and stay with the watchdog's conjunction. `docs/DriverLifecycle.md`. diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index 31d79580..ae56473f 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -458,13 +458,18 @@ alternatives: communication" — must not raise: a stalled link that resumes keeps its session, subscriptions and in-place recovery, and tearing it down would cost all three. Those stay with the watchdog, whose conjunction is the honest test for them. - **Today only `OpcUaClientDriver` implements it:** keep-alive `BadSessionIdInvalid` / `BadSessionClosed` / - `BadSessionNotActivated` / `BadServerHalted`, or a subscription the server dropped (see + `BadSessionNotActivated` / `BadServerHalted`, or a subscription the server REPORTED dropped (not the driver's + publishing-stall heuristic, which is an inference from silence and stays with the watchdog; see [OpcUaClient.md](drivers/OpcUaClient.md) §"Upstream outage and recovery" for why `BadServerHalted` is on the list — it was measured, the SDK never re-opens a halted channel). A gracefully restarted upstream therefore recovers in about one reconnect interval instead of ~100 s; a killed one still waits for the watchdog. +- ⚠️ **Known gap:** "only while `Connected`" plus "once per connection" means a raise the host receives while not yet + Connected (e.g. the protocol proves the fresh connection dead before `InitializeSucceeded` is processed) is swallowed, + and that connection falls back to the watchdog — the pre-#676 recovery time, in a narrow window. - Pinned by `DriverInstanceActorConnectionLossSignalTests` (Reconnecting within seconds while the watchdog is set to an - hour; a falsification control with no raise that stays Connected; a raise mid-connect ignored; detach on stop) and - `OpcUaClientConnectionLossSignalTests` (definitive raises once, ambiguous never, re-armed after re-init). + hour; a falsification control with no raise that stays Connected; a raise mid-connect ignored; a stray watchdog + `DisconnectObserved` in `Reconnecting` is a no-op; detach on stop) and `OpcUaClientConnectionLossSignalTests` + (definitive raises once, ambiguous and the stall heuristic never, re-armed after re-init). **The reconnect retry now tears down before re-initialising.** `Reconnecting`'s `RetryConnect` calls `ShutdownAsync` before `InitializeAsync` — the sequence every driver's own `ReinitializeAsync` already diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 38c0de5e..9c53b1a3 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -60,8 +60,9 @@ GetEndpoints returns each endpoint with the URL the **server** believes it lives container port publish or a proxy that address is unreachable from the client — or reachable but bypassing the proxy. The driver therefore rewrites the selected endpoint's **host and port** to the configured URL's whenever the schemes match, keeping the advertised path (the SDK's own `CoreClientUtils.SelectEndpoint` rule, which the driver's -policy-aware endpoint selection used to bypass). A different scheme is left alone; an unparseable advertised URL falls -back to the configured one. `OpcUaClientDriver.RewriteEndpointUrl`. +policy-aware endpoint selection used to bypass). A configured URL with **no port** keeps the advertised port (copying +the port-less `-1` would have sent the session to 4840). A different scheme is left alone; an unparseable advertised +URL falls back to the configured one. `OpcUaClientDriver.RewriteEndpointUrl`. ### Session lifetime @@ -96,17 +97,30 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal freezes a node. The outage Bad never moves the comparison; a re-initialise resets it. ⚠️ **By design, inside the window a notification whose source time EQUALS the last accepted one is dropped** — so a status-only change that keeps the old source timestamp (not what OPC UA servers are supposed to do) is lost during - the window only. When the window closes the driver logs ONE Information line — `post-recovery replay window closed — - dropped N late notifications across M refs after recovery` — because the per-drop line is Debug and the + the window only (the trailing re-read below repairs it). When the window closes — or a second recovery re-arms it — + the driver logs ONE line, `post-recovery replay window closed — dropped N late notifications across M refs after + recovery`, at **Information when N > 0** and Debug otherwise, because the per-drop line is Debug and the `stale_notification_dropped` counter is OTel-only, which made the guard unobservable on the docker-dev rig (#676). + A re-arm starts the count afresh. - **Trailing re-read after an in-place recovery (#676).** Found by the TCP-stall wire test: a change the server published just after the stall ended rode a Publish response that had been in flight across the stall; the SDK timed that request out (`BadTimeout`), discarded the response and never republished the message (the following keep-alives moved the sequence number past it), so the node stayed **Good on the pre-change value indefinitely**. `Timeout + KeepAliveInterval` after the recovery — when every request in flight across the stall has completed or - timed out — every subscribed ref is read once more, and any Good value **newer** than the last one accepted for that - ref is published (through the replay guard's mark while the window is open, so it can never regress a node). An - Information line names how many refs it had to repair. + timed out — every subscribed ref is read once more and published when its **(value, status) differs** from the last + value delivered for that ref **and is not older** than it. Any status counts: a Good→Bad/Uncertain change lost in the + pipeline is as stale as a lost value. "Older" compares source timestamps only when both carry one; a read with no + source timestamp (a server that does not send them) is the server's current state and counts as newer. An **equal** + source timestamp with different content IS published — it is the server's current state, not a replay, and it repairs + the equal-timestamp status-only change the replay guard drops by design. An identical (value, status) is never + republished. An Information line names how many refs it had to repair. The trailing read and the replay-guard disarm + timer run on the driver clock and are cancelled when the session is released (re-initialise / shutdown / dispose). +- **No older value is ever delivered after a newer one.** The three paths that publish upstream data — live + notifications, the recovery read and the trailing read — make their "may I publish this?" decision and the publish + inside one driver-wide critical section. A single lock rather than one per ref: a session's notifications arrive on + its publish thread one message at a time, so the only contention is a recovery/trailing read overlapping a + notification, and an uncontended lock costs tens of nanoseconds per notification. Subscribers run inside it — the + host's handler only `Tell`s an actor — so a subscriber must never block on, or call back into, the driver. - **The session does not come back** (an upstream restart: the server forgot the session) ⇒ the driver stays `Degraded` with its read stamp frozen, which is the #527 comms-loss watchdog's trip condition; `DriverInstanceActor` runs `ShutdownAsync` + `InitializeAsync` (retried every 10 s while the server is down) and re-subscribes data + @@ -117,7 +131,7 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal raises `ConnectionLost` — **once per session**, re-armed by a re-initialise — when the loss is certain, so the host re-initialises at once instead of after the watchdog's 90 s. Definitive means: a keep-alive status of `BadSessionIdInvalid` / `BadSessionClosed` / `BadSessionNotActivated` (the server forgot, closed or never activated - the session) or `BadServerHalted`, or a tracked subscription declared lost (below). `BadServerHalted` is on the list + the session) or `BadServerHalted`, or a subscription the **server** reported gone (below). `BadServerHalted` is on the list because it was **measured**: on the restart wire test a gracefully stopped server reports `BadServerHalted` (and the keep-alives alternate it with `BadNoCommunication`), and the SDK never re-opens a channel the server halted — 15 s after the server was back, the same session still failed every request with `BadServerHalted`; `BadSessionIdInvalid` @@ -125,7 +139,11 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal `BadRequestTimeout`/`BadTimeout` (what a stalled link produces) — is **ambiguous** (the link may resume and keep its session, see the stall test) and stays with the watchdog. Consequence: a gracefully restarted upstream recovers in roughly one reconnect interval (≈10 s) instead of ~100 s; an upstream that vanishes without a halt (killed, cable - pulled) still waits for the watchdog. + pulled) still waits for the watchdog. ⚠️ **Known gap:** the host acts on the signal only while the driver is + `Connected`, and the driver raises it once per session — so a definitive loss raised while the host is not + Connected (e.g. `BadServerHalted` on the fresh session before the host has processed the successful initialise) is + swallowed and that session falls back to the watchdog. Re-raising on later keep-alives would close it but was left out + to keep the once-per-session contract simple; the cost is the pre-#676 recovery time, in a narrow window. - **The subscription dies but the session lives** (an outage longer than the subscription lifetime — `LifetimeCount` x publishing interval, often well under the session timeout — or a server that drops it) ⇒ keep-alives stay Good, so they cannot be the signal. A `PublishError` of `BadNoSubscription` (while subscriptions are tracked) or @@ -133,8 +151,10 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal persists for one more subscription keep-alive window + 5 s across good keep-alives, declares it lost: `Degraded` (`"subscription lost: …"`), the outage published once, and — sticky until the session is released — **nothing** advances `LastSuccessfulRead` any more (not good keep-alives, not a sibling subscription's notifications), so the - watchdog's re-initialise recreates it. Both detectors are wire-tested independently. A dead subscription never comes - back on its session, so this also raises `ConnectionLost` (#676) and the re-initialise happens at once. + watchdog's re-initialise recreates it. Both detectors are wire-tested independently. A subscription the **server** + reported gone (`BadNoSubscription` / `BadSubscriptionIdInvalid`) never comes back on its session, so that also raises + `ConnectionLost` (#676) and the re-initialise happens at once; the `PublishingStopped` **heuristic** is an inference + from silence and keeps going through the watchdog. - An in-place recovery is bound to the session whose keep-alive recovered: one overtaken by a re-initialise neither publishes, nor degrades, nor re-arms the outage latch of the new session. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs index 0eb7615b..b3cebfd9 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -138,8 +138,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// /// Gitea #676 — raised at most once per session, and only on a DEFINITIVE loss - /// (, or a tracked subscription the server dropped). Re-armed when the session - /// is released (re-initialise / shutdown). + /// (, or a tracked subscription the SERVER reported dropped — never the + /// publishing-stall heuristic). Re-armed when the session is released (re-initialise / shutdown). Known gap: the host + /// acts on it only while Connected, so a raise it receives before it has processed the successful initialise is + /// swallowed and that session falls back to the #527 watchdog. public event EventHandler? ConnectionLost; // OPC UA StatusCode constants the driver surfaces for local-side faults. Upstream-server @@ -355,6 +357,26 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// Serialises arming against expiry, so an expiry racing a re-arm cannot wipe the new seeds. private readonly object _replayGuardLock = new(); + /// + /// Review of 3b88056f — serialises "decide whether this upstream value may be published" with "publish it" across + /// the three paths that publish upstream data: live notifications, the recovery read and the trailing reconcile + /// read. Without it a path could pass its newer-than check, lose the CPU to another path publishing a NEWER value, + /// and then deliver its older one last — regressing the node. One driver-wide lock rather than one per ref: SDK + /// notifications for a session arrive on its publish thread one message at a time, so the only contention is a + /// recovery/trailing read overlapping a notification (seconds after an outage), and an uncontended lock costs tens + /// of nanoseconds per notification. Subscribers are invoked under it — the host's handler only Tells an + /// actor — so a subscriber must never block on, or call back into, this driver. + /// + private readonly Lock _deliveryLock = new(); + + /// Review of 3b88056f — cancels the post-recovery background work (trailing read, replay-guard disarm timer) + /// of the current session; cancelled and replaced whenever the session is released, and on dispose. + private CancellationTokenSource _sessionWorkCts = new(); + + /// Test seam (review of 3b88056f): invoked inside the trailing read's critical section, after it decided + /// to publish a ref and before it does — where a racing notification would otherwise overtake it. + internal Action? BeforeReconcilePublishForTest { get; set; } + /// Floor of the replay-guard window; the live gate saw the late replay 4–5 s after recovery. private static readonly TimeSpan ReplayGuardMinimum = TimeSpan.FromSeconds(30); @@ -921,7 +943,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// Gitea #676 — the address to open the session on. A server reports its endpoints with ITS view of its own /// address; behind NAT, a port map, a container port publish or a proxy that address is unreachable from here /// (or bypasses the proxy). When the advertised URL's scheme matches the configured one, the configured host - /// AND port replace the advertised ones and the advertised path is kept — the SDK's own + /// AND port (when it names one) replace the advertised ones and the advertised path is kept — the SDK's own /// CoreClientUtils.SelectEndpoint rule, which this driver's policy-aware selection bypassed. A different /// scheme is a different transport and is left alone; an unparseable advertised URL falls back to the /// configured one. @@ -936,7 +958,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit || !Uri.TryCreate(advertisedUrl, UriKind.Absolute, out var advertised)) return configuredUrl; if (!string.Equals(advertised.Scheme, configured.Scheme, StringComparison.OrdinalIgnoreCase)) return advertisedUrl; - var builder = new UriBuilder(advertised) { Host = configured.DnsSafeHost, Port = configured.Port }; + // A port-less configured URL (Uri.Port == -1 for opc.tcp) keeps the advertised port — copying -1 would drop it + // and the SDK would fall back to 4840 (review of 3b88056f; same fix as the client's EndpointSelector). + var builder = new UriBuilder(advertised) + { + Host = configured.DnsSafeHost, + Port = configured.Port > 0 ? configured.Port : advertised.Port, + }; return builder.Uri.ToString(); } @@ -1042,6 +1070,16 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); // #676: re-arm the definitive-loss signal for the next session. Interlocked.Exchange(ref _connectionLostRaised, 0); + // Review of 3b88056f: the released session's post-recovery timers must not outlive it. + CancelSessionWork(); + } + + /// Cancel the current session's background work and start a fresh token for the next one. + private void CancelSessionWork() + { + var old = Interlocked.Exchange(ref _sessionWorkCts, new CancellationTokenSource()); + try { old.Cancel(); } catch (ObjectDisposedException) { } + old.Dispose(); } /// Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream. @@ -1823,27 +1861,32 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit StatusCode: dv.StatusCode.Code, SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp, ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? DateTime.UtcNow : dv.ServerTimestamp); - // F2: inside the post-recovery replay window only, a late OLDER notification must not regress the node. - if (IsLateReplay(fullRef, snapshot.SourceTimestampUtc, out var newest)) + // Review of 3b88056f: the replay check and the publish are one critical section shared with the recovery and + // trailing reads, so an older value can never be delivered after a newer one (see _deliveryLock). + lock (_deliveryLock) { - _diag.Record(OpcUaClientDiagnostics.StaleNotificationDropped); - // #676: the OTel counter and the Debug line below are invisible on a default-configured node; the - // per-recovery Information summary logged at disarm is what makes the guard observable. - lock (_replayGuardLock) + // F2: inside the post-recovery replay window only, a late OLDER notification must not regress the node. + if (IsLateReplay(fullRef, snapshot.SourceTimestampUtc, out var newest)) { - _replayDrops++; - _replayDropRefs.Add(fullRef); + _diag.Record(OpcUaClientDiagnostics.StaleNotificationDropped); + // #676: the OTel counter and the Debug line below are invisible on a default-configured node; the + // per-recovery Information summary logged at disarm is what makes the guard observable. + lock (_replayGuardLock) + { + _replayDrops++; + _replayDropRefs.Add(fullRef); + } + _logger.LogDebug( + "OpcUaClientDriver '{DriverInstanceId}': dropped a late replayed notification for {Ref} after a recovery " + + "(source {Source:O} is not newer than {Newest:O}).", _driverInstanceId, fullRef, snapshot.SourceTimestampUtc, + new DateTime(newest, DateTimeKind.Utc)); + return; } - _logger.LogDebug( - "OpcUaClientDriver '{DriverInstanceId}': dropped a late replayed notification for {Ref} after a recovery " + - "(source {Source:O} is not newer than {Newest:O}).", _driverInstanceId, fullRef, snapshot.SourceTimestampUtc, - new DateTime(newest, DateTimeKind.Utc)); - return; + RecordUpstreamGood(fullRef, snapshot); + _lastValues[fullRef] = snapshot; + _awaitingRecovery.TryRemove(fullRef, out _); + OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot)); } - RecordUpstreamGood(fullRef, snapshot); - _lastValues[fullRef] = snapshot; - _awaitingRecovery.TryRemove(fullRef, out _); - OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot)); } /// #670 F2 — remember the last accepted Good upstream source time for . @@ -1901,10 +1944,14 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit _replayDrops = 0; _replayDropRefs.Clear(); } - _logger.LogInformation( + LogReplaySummary(drops, refs); + } + + /// One line per replay window: Information when the guard dropped something, Debug otherwise. + private void LogReplaySummary(int drops, int refs) => + _logger.Log(drops > 0 ? LogLevel.Information : LogLevel.Debug, "OpcUaClientDriver '{DriverInstanceId}': post-recovery replay window closed — dropped {Drops} late " + "notifications across {Refs} refs after recovery.", _driverInstanceId, drops, refs); - } /// /// #670 F2 — arm the replay guard on a same-session recovery: every ref with a known Good upstream source time, @@ -1919,12 +1966,19 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit var keepAliveWindow = TimeSpan.FromMilliseconds(sub.CurrentPublishingInterval * (double)sub.CurrentKeepAliveCount); if (keepAliveWindow * 2 > window) window = keepAliveWindow * 2; } + int supersededDrops, supersededRefs; lock (_replayGuardLock) { + // A re-arm inside an open window ends that window: report what it dropped and count afresh. + supersededDrops = _replayDrops; + supersededRefs = _replayDropRefs.Count; + _replayDrops = 0; + _replayDropRefs.Clear(); _replayGuard.Clear(); foreach (var kv in _lastGoodSourceTicks) _replayGuard[kv.Key] = kv.Value; Interlocked.Exchange(ref _replayGuardUntilTicks, (now + window).Ticks); } + if (supersededDrops > 0) LogReplaySummary(supersededDrops, supersededRefs); return (now + window).Ticks; } @@ -1932,14 +1986,15 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// #676 — disarm the replay guard (and log its summary) when its window ends, even if no notification arrives to /// disarm it lazily in — a quiet subscription would otherwise never log the summary. /// - private async Task DisarmReplayGuardWhenDueAsync(long until) + private async Task DisarmReplayGuardWhenDueAsync(long until, CancellationToken ct) { try { var wait = new DateTime(until, DateTimeKind.Utc) - Clock.GetUtcNow().UtcDateTime; - if (wait > TimeSpan.Zero) await Task.Delay(wait + TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); + if (wait > TimeSpan.Zero) await Task.Delay(wait + TimeSpan.FromMilliseconds(100), Clock, ct).ConfigureAwait(false); if (Clock.GetUtcNow().UtcDateTime.Ticks >= until) DisarmReplayGuard(until); } + catch (OperationCanceledException) when (ct.IsCancellationRequested) { /* session released */ } catch (Exception ex) { _logger.LogDebug(ex, "OpcUaClientDriver '{DriverInstanceId}': replay-guard disarm timer faulted.", _driverInstanceId); @@ -2820,14 +2875,16 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit "re-reading the refs published Bad during the outage.", _driverInstanceId); var recovery = Task.Run(() => RecoverAfterOutageAsync(sender)); _recoveryTask = recovery; - _reconcileTask = ReconcileAfterRecoveryAsync(sender, recovery, PostRecoveryReconcileDelay()); - _ = DisarmReplayGuardWhenDueAsync(guardUntil); + var sessionWork = _sessionWorkCts.Token; + _reconcileTask = ReconcileAfterRecoveryAsync(sender, recovery, PostRecoveryReconcileDelay(), sessionWork); + _ = DisarmReplayGuardWhenDueAsync(guardUntil, sessionWork); return; } if (FindStalledSubscription(now) is { } stalled) { - DeclareSubscriptionLost($"subscription '{stalled}' stopped publishing while the session stayed up", now); + DeclareSubscriptionLost($"subscription '{stalled}' stopped publishing while the session stayed up", now, + serverReported: false); return; } @@ -2923,7 +2980,12 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// frozen (sticky until the session is released), and the outage published once, so the host's watchdog /// re-initialises and re-subscribes. /// - private void DeclareSubscriptionLost(string reason, DateTime now) + /// Why, for the log and health. + /// Detection time. + /// True when the SERVER said the subscription is gone (BadNoSubscription / + /// BadSubscriptionIdInvalid) — definitive, so is raised. False for the publishing-stall + /// heuristic, an inference from silence that stays with the #527 watchdog (review of 3b88056f). + private void DeclareSubscriptionLost(string reason, DateTime now, bool serverReported) { if (Interlocked.Exchange(ref _subscriptionLost, 1) == 1) return; UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"subscription lost: {reason}")); @@ -2931,9 +2993,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit "OpcUaClientDriver '{DriverInstanceId}': {Reason}; subscribed values are BadNoCommunication until the " + "host re-initialises the driver and re-subscribes.", _driverInstanceId, reason); if (Volatile.Read(ref _sessionLost) == 0) PublishCommsLoss(now); - // #676: a dead subscription never comes back on this session — only the re-subscribe after a re-initialise - // recreates it — so this is definitive: ask the host now rather than after the watchdog's 90 s. - RaiseConnectionLost($"subscription lost: {reason}"); + // #676: a subscription the SERVER says is gone never comes back on this session — only the re-subscribe after a + // re-initialise recreates it — so ask the host now rather than after the watchdog's 90 s. The stall heuristic is + // an inference, not a server statement, and keeps going through the watchdog. + if (serverReported) RaiseConnectionLost($"subscription lost: {reason}"); } /// @@ -2956,7 +3019,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit else return; if (tracked) - DeclareSubscriptionLost($"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime); + DeclareSubscriptionLost($"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime, + serverReported: true); } /// @@ -3011,15 +3075,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit for (var i = 0; i < refs.Count; i++) { if (!Current()) return; - if (!_awaitingRecovery.TryRemove(refs[i], out _)) continue; // a notification got there first - _lastValues[refs[i]] = results[i]; - RecordUpstreamGood(refs[i], results[i]); - // The read is upstream data published inside the replay window: a replay older than it is late too. - if (results[i].SourceTimestampUtc is { } readAt && StatusCode.IsGood(results[i].StatusCode)) - _replayGuard.AddOrUpdate(refs[i], readAt.Ticks, (_, prior) => Math.Max(prior, readAt.Ticks)); - foreach (var rs in _subscriptions.Values) - if (rs.ItemHandlers.Any(p => RefOf(p.Item) == refs[i])) - RaiseDataChange(new DataChangeEventArgs(rs.Handle, refs[i], results[i])); + // One critical section with the live notification path (_deliveryLock): a notification that + // removed the ref first wins, and none can overtake this publish once it has decided. + lock (_deliveryLock) + { + if (!_awaitingRecovery.TryRemove(refs[i], out _)) continue; // a notification got there first + PublishUpstreamLocked(refs[i], results[i]); + } } } @@ -3035,49 +3097,68 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } } + /// + /// Publish an upstream value read (not notified) for to every subscription carrying it, + /// recording it as the last delivered value and — when Good with a source time — as the replay guard's mark, so a + /// replay older than the read is late too. Caller holds . + /// + private void PublishUpstreamLocked(string fullRef, DataValueSnapshot value) + { + _lastValues[fullRef] = value; + RecordUpstreamGood(fullRef, value); + if (value.SourceTimestampUtc is { } readAt && StatusCode.IsGood(value.StatusCode) + && Interlocked.Read(ref _replayGuardUntilTicks) != 0) + _replayGuard.AddOrUpdate(fullRef, readAt.Ticks, (_, prior) => Math.Max(prior, readAt.Ticks)); + foreach (var rs in _subscriptions.Values) + if (rs.ItemHandlers.Any(p => RefOf(p.Item) == fullRef)) + RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, value)); + } + /// /// #676 — the trailing reconcile read after an in-place recovery. Found by the TCP-stall wire test: a value change /// the server published just after a stall ended rode a Publish response that had been in flight across the /// stall; the SDK timed that request out (BadTimeout), discarded the response, and never republished the - /// message (the following keep-alives advanced the sequence number past it). The node stayed Good on the value - /// from the recovery read — stale, indefinitely, for a value that does not change again. Once every request that - /// was in flight across the stall has completed or timed out (), every - /// subscribed ref is read once more and any Good value NEWER than the last one accepted for that ref is published - /// (through the replay guard's mark while its window is open, so it can never regress a node). + /// message (the following keep-alives advanced the sequence number past it). The node stayed on the value from + /// the recovery read — stale, indefinitely, for a value that does not change again. Once every request that was + /// in flight across the stall has completed or timed out (), every + /// subscribed ref is read once more and published when says the read differs from + /// what was last delivered and is not older than it. The decision and the publish are one critical section with + /// the live notification path (), so a newer notification can never be overtaken. + /// Bound to the session's work token: a re-initialise or shutdown cancels it. /// /// The session that recovered; nothing happens if it is no longer current or is in outage. /// The recovery read, which runs first so this read is compared against its result. - /// How long to wait before reading. - private async Task ReconcileAfterRecoveryAsync(ISession session, Task recovery, TimeSpan delay) + /// How long to wait before reading (on ). + /// The session's background-work token. + private async Task ReconcileAfterRecoveryAsync(ISession session, Task recovery, TimeSpan delay, CancellationToken ct) { - bool Current() => ReferenceEquals(session, Session) + bool Current() => !ct.IsCancellationRequested + && ReferenceEquals(session, Session) && Volatile.Read(ref _sessionLost) == 0 && Volatile.Read(ref _subscriptionLost) == 0; try { - if (delay > TimeSpan.Zero) await Task.Delay(delay).ConfigureAwait(false); - await recovery.ConfigureAwait(false); + if (delay > TimeSpan.Zero) await Task.Delay(delay, Clock, ct).ConfigureAwait(false); + await recovery.WaitAsync(ct).ConfigureAwait(false); if (!Current()) return; var refs = _subscriptions.Values .SelectMany(rs => rs.ItemHandlers.Select(p => RefOf(p.Item))) .Distinct(StringComparer.Ordinal) .ToList(); if (refs.Count == 0) return; - var (results, transportOk) = await ReadCoreAsync(refs, CancellationToken.None, session).ConfigureAwait(false); + var (results, transportOk) = await ReadCoreAsync(refs, ct, session).ConfigureAwait(false); if (!transportOk) return; // a new outage — the keep-alive reports it var published = 0; for (var i = 0; i < refs.Count; i++) { if (!Current()) return; - var r = results[i]; - if (!StatusCode.IsGood(r.StatusCode) || r.SourceTimestampUtc is not { } src) continue; - if (!IsNewerThanAccepted(refs[i], src)) continue; - RecordUpstreamGood(refs[i], r); - _lastValues[refs[i]] = r; - _awaitingRecovery.TryRemove(refs[i], out _); - foreach (var rs in _subscriptions.Values) - if (rs.ItemHandlers.Any(p => RefOf(p.Item) == refs[i])) - RaiseDataChange(new DataChangeEventArgs(rs.Handle, refs[i], r)); + lock (_deliveryLock) + { + if (!IsReconcileRepair(refs[i], results[i])) continue; + BeforeReconcilePublishForTest?.Invoke(refs[i]); + _awaitingRecovery.TryRemove(refs[i], out _); + PublishUpstreamLocked(refs[i], results[i]); + } published++; } if (published > 0) @@ -3086,6 +3167,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit "whose latest change never arrived through the subscription; published them.", _driverInstanceId, published, refs.Count); } + catch (OperationCanceledException) when (ct.IsCancellationRequested) { /* session released */ } catch (Exception ex) { _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': trailing re-read after a recovery failed.", @@ -3093,14 +3175,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } } - /// #676 — whether a Good upstream value with source time is newer than the last one - /// accepted for . Inside the replay window this IS the guard's check (and advances its mark); - /// outside it, a strict comparison against the last accepted Good source time. - private bool IsNewerThanAccepted(string fullRef, DateTime source) + /// + /// Whether the trailing read's repairs : its (value, status) + /// differs from the last value delivered for the ref, and it is not OLDER than it. Any status counts — a Good→Bad + /// or Good→Uncertain change lost in the stalled pipeline is as stale as a lost value. "Older" is judged on source + /// timestamps only when both carry one; a read with no source timestamp (a server that does not send them) is the + /// server's CURRENT state, read after the stalled pipeline drained, so it counts as newer. An EQUAL source + /// timestamp with different content is published too: it is not a replay but the server's current state, and it + /// repairs the one change the replay guard drops by design (a status-only change that kept its source time inside + /// the window). Caller holds . + /// + private bool IsReconcileRepair(string fullRef, DataValueSnapshot read) { - var until = Interlocked.Read(ref _replayGuardUntilTicks); - if (until != 0 && Clock.GetUtcNow().UtcDateTime.Ticks < until) return !IsLateReplay(fullRef, source, out _); - return !_lastGoodSourceTicks.TryGetValue(fullRef, out var last) || source.Ticks > last; + if (!_lastValues.TryGetValue(fullRef, out var last)) return true; + if (last.StatusCode == read.StatusCode && Utils.IsEqual(last.Value, read.Value)) return false; + return read.SourceTimestampUtc is not { } src + || last.SourceTimestampUtc is not { } lastSrc + || src >= lastSrc; } /// Quality-only annotation (#626 pattern) of every subscribed alarm condition. @@ -3204,6 +3295,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } catch { /* timeout or already disposed — proceed */ } _gate.Dispose(); + try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { } SdkTelemetry.Release(_sdkTelemetry); } @@ -3233,6 +3325,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } catch { /* timeout or already disposed */ } _gate.Dispose(); + try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { } SdkTelemetry.Release(_sdkTelemetry); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 6940964a..faadafff 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -512,8 +512,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(msg => HandleDisconnectObserved(msg.Reason)); // #676 — the driver KNOWS its connection is gone (e.g. OpcUaClient: the upstream server halted or forgot the // session). Same handling as the watchdog's disconnect, minus the ~90 s of silence it would need to prove it. - // Handled inline rather than re-told as DisconnectObserved, so it cannot land in Reconnecting as an unhandled - // message. + // Handled inline rather than re-told as DisconnectObserved (Reconnecting also swallows a stray DisconnectObserved). Receive(msg => HandleDisconnectObserved($"driver reported connection lost: {msg.Args.Reason}")); Receive(_ => { @@ -771,6 +770,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(StoreDesiredSubscriptions); Receive(_ => { /* already reconnecting — no-op */ }); Receive(IgnoreConnectionLoss); + // A watchdog DisconnectObserved self-sent on the Connected tick can still be queued when a ConnectionLossRaised + // (or another trip) has already moved us here — already reconnecting, so it is a no-op, not an Unhandled. + Receive(_ => { /* already reconnecting — no-op */ }); // ResubscribeDesired self-Tells Subscribe; HandleSubscribeAsync replies SubscriptionEstablished to the // sender, which on the self-resubscribe path is Self. Swallow it (trace only) so it doesn't dead-letter. Receive(msg => @@ -1108,10 +1110,6 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers _rediscoveryHandler = null; } - /// Subscribe the driver's (if it is - /// one), marshaling each transition to the actor thread. Idempotent; mirrors - /// , including the PreStart-not-per-connect placement — a probe - /// loop can report a host down while the driver itself is between connects. /// /// Subscribe the driver's (#676), if it is one, /// marshalling each raise to the actor thread — actor state is never touched on the driver's thread. Attached @@ -1156,6 +1154,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers PublishHealthSnapshot(); } + /// Subscribe the driver's (if it is + /// one), marshaling each transition to the actor thread. Idempotent; mirrors + /// , including the PreStart-not-per-connect placement — a probe + /// loop can report a host down while the driver itself is between connects. private void AttachHostStatusSource() { if (_driver is not IHostConnectivityProbe src || _hostStatusHandler is not null) return; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs index 109074c0..e3f34208 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs @@ -38,10 +38,12 @@ public sealed class OpcUaClientConnectionLossSignalTests lock (Entries) Entries.Add((logLevel, formatter(state, exception))); } - public string[] Info(string contains) + public string[] Info(string contains) => At(LogLevel.Information, contains); + + public string[] At(LogLevel level, string contains) { lock (Entries) - return [.. Entries.Where(e => e.Level == LogLevel.Information && e.Message.Contains(contains)).Select(e => e.Message)]; + return [.. Entries.Where(e => e.Level == level && e.Message.Contains(contains)).Select(e => e.Message)]; } } @@ -231,6 +233,31 @@ public sealed class OpcUaClientConnectionLossSignalTests drv.GetHealth().State.ShouldBe(DriverState.Degraded); } + /// + /// Review of 3b88056f: only a SERVER-reported subscription loss is definitive. The publishing-stall heuristic + /// (SDK PublishingStopped persisting past a grace) is an inference from silence — it still degrades health + /// and freezes the stamp, but recovery stays with the #527 watchdog. + /// + [Fact] + public void The_publishing_stall_heuristic_does_not_raise_ConnectionLost() + { + var (drv, _, clock, _) = NewConnectedDriver(); + using var _ = drv; + drv.AddSubscriptionForTest([SpeedRef]); + drv.IsPublishingStopped = _ => true; + drv.PublishingStallGrace = _ => TimeSpan.FromSeconds(10); + var raised = 0; + drv.ConnectionLost += (_, _) => raised++; + + drv.RaiseKeepAliveForTest(StatusCodes.Good); + clock.Advance(TimeSpan.FromSeconds(11)); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + + // Positive control: the heuristic DID fire. + drv.GetHealth().LastError.ShouldNotBeNull().ShouldContain("subscription lost"); + raised.ShouldBe(0); + } + // ── Replay-drop summary ────────────────────────────────────────────────────────────────────────────────────── [Fact] @@ -264,7 +291,28 @@ public sealed class OpcUaClientConnectionLossSignalTests } [Fact] - public async Task A_window_with_no_drops_still_reports_it_closed_and_counts_restart_per_window() + public async Task A_window_with_no_drops_closes_at_Debug_not_Information() + { + var (drv, session, clock, log) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.FromDays(1); + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0)); + SetupRead(session, () => t0, 1); + + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + clock.Advance(TimeSpan.FromSeconds(31)); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(2, t0.AddSeconds(40))); + + log.Info("replay window closed").ShouldBeEmpty("nothing was dropped — not worth an Information line per recovery"); + log.At(LogLevel.Debug, "replay window closed").ShouldHaveSingleItem().ShouldContain("dropped 0"); + } + + [Fact] + public async Task A_re_arm_inside_an_open_window_reports_the_earlier_drops_once_and_starts_counting_afresh() { var (drv, session, clock, log) = NewConnectedDriver(); using var _ = drv; @@ -278,19 +326,18 @@ public sealed class OpcUaClientConnectionLossSignalTests drv.RaiseKeepAliveForTest(StatusCodes.Good); await drv.RecoveryTaskForTest; drv.DeliverNotificationForTest(handle, SpeedRef, Good(0, t0.AddSeconds(-1))); // one drop in window 1 - clock.Advance(TimeSpan.FromSeconds(31)); - drv.DeliverNotificationForTest(handle, SpeedRef, Good(2, t0.AddSeconds(40))); - drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + clock.Advance(TimeSpan.FromSeconds(5)); // window 1 still open … + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); // … a second outage re-arms it drv.RaiseKeepAliveForTest(StatusCodes.Good); await drv.RecoveryTaskForTest; clock.Advance(TimeSpan.FromSeconds(31)); drv.DeliverNotificationForTest(handle, SpeedRef, Good(3, t0.AddSeconds(80))); - log.Info("replay window closed").ShouldBe([ - "OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 1 late notifications across 1 refs after recovery.", - "OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 0 late notifications across 0 refs after recovery.", - ]); + log.Info("replay window closed").ShouldBe( + ["OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 1 late notifications across 1 refs after recovery."], + "window 1's drop is reported once, when the re-arm supersedes it — not carried into window 2's summary"); + log.At(LogLevel.Debug, "replay window closed").ShouldHaveSingleItem().ShouldContain("dropped 0"); } // ── Trailing reconcile read ────────────────────────────────────────────────────────────────────────────────── @@ -405,4 +452,166 @@ public sealed class OpcUaClientConnectionLossSignalTests It.IsAny(), It.IsAny()), Times.Once, "only the recovery read — the trailing read must not run against a session that is no longer current"); } + + /// Wires the mock session so the first read (the recovery read) returns and + /// every later one (the trailing read) returns . + private static Func SetupRecoveryThenTrailing(Mock session, DataValue recovery, DataValue trailing) + { + var reads = 0; + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(() => new ReadResponse + { + ResponseHeader = new ResponseHeader(), + Results = [Interlocked.Increment(ref reads) > 1 ? trailing : recovery], + DiagnosticInfos = [], + }); + return () => Volatile.Read(ref reads); + } + + private static async Task> RecoverAndReconcileAsync( + OpcUaClientDriver drv, string fullRef = SpeedRef) + { + var seen = new List<(uint Status, object? Value)>(); + drv.OnDataChange += (_, e) => { if (e.FullReference == fullRef) lock (seen) seen.Add((e.Snapshot.StatusCode, e.Snapshot.Value)); }; + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + await drv.ReconcileTaskForTest; + return seen; + } + + /// Review of 3b88056f: a Good→Bad change lost in the stalled pipeline left the node stale-GOOD; the + /// trailing read repairs any (value, status) difference, not just Good ones. + [Fact] + public async Task The_trailing_re_read_repairs_a_lost_change_to_Bad() + { + var (drv, session, clock, _) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0)); + var reads = SetupRecoveryThenTrailing(session, Good(10f, t0), + new DataValue(new Variant(10f), StatusCodes.BadSensorFailure, t0.AddSeconds(9), t0.AddSeconds(9))); + + var seen = await RecoverAndReconcileAsync(drv); + + reads().ShouldBe(2); + seen.ShouldBe([(BadNoCommunication, (object?)10f), (0u, 10f), ((uint)StatusCodes.BadSensorFailure, 10f)]); + } + + /// A server that sends no source timestamp: the re-read is the server's CURRENT state, read after the + /// stalled pipeline drained, so a differing value is newer than anything delivered before — it is published. + [Fact] + public async Task The_trailing_re_read_repairs_a_server_that_sends_no_source_timestamp() + { + var (drv, session, clock, _) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + DataValue NoSource(object v, DateTime server) => + new(new Variant(v), StatusCodes.Good, DateTime.MinValue, server); + drv.DeliverNotificationForTest(handle, SpeedRef, NoSource(10f, t0)); + var reads = SetupRecoveryThenTrailing(session, NoSource(10f, t0.AddSeconds(1)), NoSource(40f, t0.AddSeconds(9))); + + var seen = await RecoverAndReconcileAsync(drv); + + reads().ShouldBe(2); + seen.ShouldBe([(BadNoCommunication, (object?)10f), (0u, 10f), (0u, 40f)]); + } + + /// + /// Equal source timestamp, different status: the re-read is the server's current state, not a replay, so it is + /// published — which also repairs the one case the replay guard drops by design (a status-only change that kept + /// its source timestamp inside the window). An IDENTICAL (value, status) is not republished. + /// + [Fact] + public async Task The_trailing_re_read_publishes_a_status_only_change_at_an_equal_source_time_but_not_an_identical_value() + { + var (drv, session, clock, _) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero; + var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0)); + drv.DeliverNotificationForTest(handle, CounterRef, Good(7, t0)); + var reads = 0; + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(() => new ReadResponse + { + ResponseHeader = new ResponseHeader(), + Results = Interlocked.Increment(ref reads) > 1 + ? [new DataValue(new Variant(10f), StatusCodes.UncertainLastUsableValue, t0, t0), Good(7, t0)] + : [Good(10f, t0), Good(7, t0)], + DiagnosticInfos = [], + }); + var seen = new List<(string Ref, uint Status)>(); + drv.OnDataChange += (_, e) => { lock (seen) seen.Add((e.FullReference, e.Snapshot.StatusCode)); }; + + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + await drv.ReconcileTaskForTest; + + reads.ShouldBe(2); + seen.Where(e => e.Status != BadNoCommunication).ShouldBe([ + (SpeedRef, 0u), (CounterRef, 0u), // recovery read + (SpeedRef, (uint)StatusCodes.UncertainLastUsableValue), // trailing read: status-only change at t0 + ]); + } + + /// + /// Review of 3b88056f: the trailing read's compare and its publish are one critical section with the live + /// notification path. Here a NEWER notification arrives exactly between the trailing read's decision and its + /// publish; it must wait, so the older re-read can never land after it and regress the node. + /// + [Fact] + public async Task A_notification_racing_the_trailing_publish_is_serialised_after_it() + { + var (drv, session, clock, _) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0)); + SetupRecoveryThenTrailing(session, Good(10f, t0), Good(40f, t0.AddSeconds(5))); + Task? racer = null; + drv.BeforeReconcilePublishForTest = _ => + { + racer = Task.Run(() => drv.DeliverNotificationForTest(handle, SpeedRef, Good(50f, t0.AddSeconds(8)))); + // Bounded: with the critical section the racer cannot finish until this publish is done. + racer.Wait(TimeSpan.FromMilliseconds(300)); + }; + + var seen = await RecoverAndReconcileAsync(drv); + await racer!.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + lock (seen) seen.Select(e => e.Value).ShouldBe([10f, 10f, 40f, 50f], "the newer 50 must be the last value delivered"); + } + + /// Review of 3b88056f: the background timers are bound to the session — a shutdown cancels a pending + /// trailing read instead of leaving it to fire (and read) against whatever session exists later. + [Fact] + public async Task Shutdown_cancels_a_pending_trailing_read() + { + var (drv, session, clock, _) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.FromHours(1); + var handle = drv.AddSubscriptionForTest([SpeedRef]); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, clock.Now.UtcDateTime)); + SetupRead(session, () => clock.Now.UtcDateTime, 10f); + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + drv.ReconcileTaskForTest.IsCompleted.ShouldBeFalse("pending for an hour"); + + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + + await drv.ReconcileTaskForTest.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientEndpointRewriteTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientEndpointRewriteTests.cs index 55a3762a..6e9ce0e8 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientEndpointRewriteTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientEndpointRewriteTests.cs @@ -34,6 +34,22 @@ public sealed class OpcUaClientEndpointRewriteTests .ShouldBe("opc.tcp://127.0.0.1:61000/OtOpcUa"); } + [Fact] + public void A_configured_url_without_a_port_keeps_the_advertised_port() + { + // Review of 3b88056f: Uri.Port is -1 for a port-less opc.tcp URL; copying it would drop the advertised port and + // the SDK would fall back to 4840. + OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://internal-host:62541/Server", "opc.tcp://gateway/Server") + .ShouldBe("opc.tcp://gateway:62541/Server"); + } + + [Fact] + public void An_IPv6_configured_host_is_bracketed() + { + OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://internal-host:4840/OtOpcUa", "opc.tcp://[fd00::1]:54840/OtOpcUa") + .ShouldBe("opc.tcp://[fd00::1]:54840/OtOpcUa"); + } + [Fact] public void A_different_scheme_is_left_alone() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverStallWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverStallWireTests.cs index 5cee9ec5..82504d64 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverStallWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverStallWireTests.cs @@ -44,9 +44,12 @@ public sealed class OpcUaClientDriverStallWireTests await using var proxy = TcpStallProxy.Start(new Uri(rig.EndpointUrl).Port); OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f); - // A 3 s operation timeout keeps the stall short: a keep-alive is declared failed when its request times out. + // A 3 s operation timeout keeps the stall short (a keep-alive is declared failed when its request times out); a + // 120 s session timeout and a 3000-cycle (300 s at 100 ms) subscription lifetime keep both far outside the + // stall, so the test exercises the in-place recovery and nothing else. await using var drv = OpcUaClientDriverWireTests.NewDriver( - rig, "wire-stall", endpointUrl: proxy.Through(rig.EndpointUrl), timeout: TimeSpan.FromSeconds(3)); + rig, "wire-stall", endpointUrl: proxy.Through(rig.EndpointUrl), timeout: TimeSpan.FromSeconds(3), + sessionTimeout: TimeSpan.FromSeconds(120), subscriptionLifetimeCount: 3000); var connectionLost = 0; drv.ConnectionLost += (_, _) => Interlocked.Increment(ref connectionLost); await drv.InitializeAsync("{}", ct); @@ -63,10 +66,19 @@ public sealed class OpcUaClientDriverStallWireTests var session = drv.Session.ShouldNotBeNull(); var connectionsBefore = proxy.AcceptedConnections; + // Bad keep-alive TICKS: the SDK reports a failed tick as two events at the same instant (the request's status and + // BadNoCommunication), so events closer than half a keep-alive interval are one tick. var badKeepAlives = 0; + var lastBadTick = DateTime.MinValue; session.KeepAlive += (_, e) => { - if (ServiceResult.IsBad(e.Status)) Interlocked.Increment(ref badKeepAlives); + if (!ServiceResult.IsBad(e.Status)) return; + lock (seen) + { + if (DateTime.UtcNow - lastBadTick < TimeSpan.FromMilliseconds(500)) return; + lastBadTick = DateTime.UtcNow; + badKeepAlives++; + } }; // The SDK's late backlog flush: a data-carrying NotificationMessage whose sequence number is BELOW one already // seen (keep-alive messages carry the next number). It is the positive control for the no-regression check. @@ -97,9 +109,10 @@ public sealed class OpcUaClientDriverStallWireTests await Task.Delay(300, ct); // > one publishing interval, so the server queues 20.0 as its own notification OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f); - // Stays Bad and frozen across a further failed keep-alive (counted, not slept for). - var badAtDetection = Volatile.Read(ref badKeepAlives); - await WaitAsync(() => Volatile.Read(ref badKeepAlives) >= badAtDetection + 2, "further keep-alives must fail while stalled"); + // Stays Bad and frozen across one further failed keep-alive tick (counted, not slept for). + int badAtDetection; + lock (seen) badAtDetection = badKeepAlives; + await WaitAsync(() => { lock (seen) return badKeepAlives > badAtDetection; }, "a further keep-alive must fail while stalled"); drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen, "the read stamp is the #527 watchdog's silence measure"); lock (seen) seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1, "one Bad per outage, not per keep-alive"); @@ -118,13 +131,15 @@ public sealed class OpcUaClientDriverStallWireTests // being answered or timed out — must still reach the node. OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 40.0f); await WaitAsync(() => { lock (seen) return seen.Any(s => s is { Status: 0u, Value: 40.0f }); }, - "a change published right after recovery must reach the node (via the subscription or the trailing re-read)"); + "a change published right after recovery must reach the node (via the subscription or the trailing re-read)", + TimeSpan.FromSeconds(60)); await drv.ReconcileTaskForTest; // The SDK flushes the notifications held during the stall seconds after recovery (the #670 F2 shape). Wait for // that flush before judging ordering, or the check below would pass on a run where nothing late arrived. await WaitAsync(() => { lock (seen) return lateMessages > 0; }, - "the SDK's late backlog flush did not arrive — the no-regression check below would be vacuous"); + "the SDK's late backlog flush did not arrive — the no-regression check below would be vacuous", + TimeSpan.FromSeconds(60)); lock (seen) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs index 7b5709f1..fc8f2a01 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs @@ -41,7 +41,8 @@ public sealed class OpcUaClientDriverWireTests internal static OpcUaClientDriver NewDriver( ServerEventRig rig, string id, string? browseRoot = null, IReadOnlyList? extraTags = null, - string? endpointUrl = null, TimeSpan? timeout = null) => + string? endpointUrl = null, TimeSpan? timeout = null, TimeSpan? sessionTimeout = null, + uint? subscriptionLifetimeCount = null) => new(new OpcUaClientDriverOptions { EndpointUrl = endpointUrl ?? rig.EndpointUrl, @@ -59,7 +60,8 @@ public sealed class OpcUaClientDriverWireTests Password = ServerEventRig.KeyPassword, AutoAcceptCertificates = true, Timeout = timeout ?? TimeSpan.FromSeconds(10), - SessionTimeout = TimeSpan.FromSeconds(30), + SessionTimeout = sessionTimeout ?? TimeSpan.FromSeconds(30), + SubscriptionLifetimeCount = subscriptionLifetimeCount ?? 1000, KeepAliveInterval = TimeSpan.FromSeconds(1), BrowseRoot = browseRoot, }, id); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConnectionLossSignalTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConnectionLossSignalTests.cs index e7da8d01..17d0c2f7 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConnectionLossSignalTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConnectionLossSignalTests.cs @@ -1,4 +1,5 @@ using Akka.Actor; +using Akka.Event; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; @@ -126,6 +127,23 @@ public sealed class DriverInstanceActorConnectionLossSignalTests : RuntimeActorT driver.InitializeCount.ShouldBe(1); } + /// Review of 3b88056f: a watchdog DisconnectObserved still queued when the connection-loss signal has + /// already moved the actor to Reconnecting is a no-op there, not an Unhandled message. + [Fact] + public async Task A_stray_DisconnectObserved_while_reconnecting_is_not_unhandled() + { + var driver = new LossSourceStubDriver(); + var (actor, parent) = Connect(driver); + var unhandled = CreateTestProbe(); + Sys.EventStream.Subscribe(unhandled.Ref, typeof(UnhandledMessage)); + await driver.RaiseAsync("keep-alive reported BadServerHalted"); + parent.ExpectMsg(TimeSpan.FromSeconds(3)).Connected.ShouldBeFalse(); + + actor.Tell(new DriverInstanceActor.DisconnectObserved("late watchdog trip")); + + unhandled.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + } + [Fact] public void The_handler_is_attached_once_and_detached_when_the_actor_stops() {