diff --git a/CLAUDE.md b/CLAUDE.md index 32e4a6e2..33452941 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -321,6 +321,12 @@ non-`Healthy` state **and** its `LastSuccessfulRead` has been frozen for 90 s. T `DefaultReconnectInterval`. Unlike the #523 drift detector this generates **no** traffic on a healthy system — it is a local `GetHealth()` read, and only reaches the network once a driver has already stopped 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 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`. - **What comms loss now does downstream — one contract, six seams (archreview R3-02).** "A value consumer never reports a quality better than its source's" is stated once in `docs/DriverLifecycle.md` §"The diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index df8ab2ec..ae56473f 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -441,6 +441,36 @@ Two consequences worth knowing: **notices**, so a silent-success termination path — a stream that ends cleanly, a poll loop that exits without throwing — bypasses it entirely. +**A driver that KNOWS its connection is gone can skip the wait: `IDriverConnectionLossSource` (Gitea #676).** The +watchdog has to prove a loss from silence, so it costs ~90 s even when the protocol already said "this connection +will never come back". An opt-in `Core.Abstractions` interface lets a driver say so: +`event EventHandler ConnectionLost` (a reason string). `DriverInstanceActor` attaches +in `PreStart` and detaches in `PostStop` — the same lifetime as `IRediscoverable`, for the same reason (the `IDriver` +outlives the actor on a respawn) — marshals each raise to the actor thread (`Self.Tell`; actor state is never touched +on the driver's thread), and acts on it **only while `Connected`**, through the very handler the watchdog's +`DisconnectObserved` uses (detach, record a fault, `Reconnecting`, teardown-then-init on the retry). In +`Connecting`/`Reconnecting`/`Stubbed` it is ignored, not deferred: a re-initialise is already pending, and acting on a +stale raise later could tear down the connection that attempt is building. The two are complementary, not +alternatives: + +- **The signal is for DEFINITIVE losses only.** A driver must raise only when no in-place recovery is possible, at most + once per connection, re-armed by a re-initialise. An ambiguous failure — a timeout, a stalled socket, "no + 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 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; 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 uses — because several drivers **overwrite** an `IAsyncDisposable` transport field inside `InitializeAsync` diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 59cf4fa2..6c1f6da7 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -50,7 +50,19 @@ provides ordered **failover** — the driver tries each candidate in turn at eve 4840 / 4841). See `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs` for every field (security policy/mode, auth type, session timeout, keep-alive, -browse root, node/depth caps). `ReconnectPeriod` is still parsed but no longer used (#670). +browse root, node/depth caps). `ReconnectPeriod` was **deleted** (#676 — unused since #670 removed the SDK reconnect +handler, and the AdminUI no longer shows it); stored configs that still carry `reconnectPeriod` keep parsing in the +factory, the Test-Connect probe and the browse picker (unknown keys are skipped), and the form drops it on the next save. + +### Endpoint address: the configured host and port win (#676) + +GetEndpoints returns each endpoint with the URL the **server** believes it lives at. Behind NAT, a port map, a +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 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 @@ -83,12 +95,57 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal without moving the mark. Outside that window every notification is accepted whatever its timestamp, so an upstream clock that steps backwards (NTP, a PLC clock reset) never 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 (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 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. A ref that received ANY delivery while the read was in flight is skipped (a per-ref delivery counter is + captured before the read is issued): without source timestamps the read's sample cannot be ordered against that + notification, and publishing it would regress the node. 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 paths that publish upstream data — live notifications, + the recovery read, the trailing read and the comms-loss Bad fan — make their "may I publish this?" decision and the + publish inside one driver-wide critical section. A single lock rather than one per ref: it also serialises delivery + across subscriptions (the SDK dispatches each subscription's messages on its own worker), which is cheap because each + section is a dictionary update plus a subscriber `Tell` and nothing awaits under it. 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 exactly the #527 comms-loss watchdog's trip condition. After 90 s - `DriverInstanceActor` runs `ShutdownAsync` + `InitializeAsync` (retried every 10 s while the server is down) and - re-subscribes data + alarms; the new subscription's initial values come back Good. Alarm last-told state survives - the re-initialise (#619), so the post-subscribe `ConditionRefresh` raises no un-ack flicker; the host's reconnect - annotation restores condition quality. + `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 + + alarms; the new subscription's initial values come back Good. Alarm last-told state survives the re-initialise + (#619), so the post-subscribe `ConditionRefresh` raises no un-ack flicker; the host's reconnect annotation restores + condition quality. +- **Fast recovery for a DEFINITIVELY lost session (#676).** The driver implements `IDriverConnectionLossSource` and + 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 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` + never arrived. Everything else — `BadNoCommunication`, `BadConnectionClosed`, `BadNotConnected`, + `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. ⚠️ **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 @@ -96,7 +153,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. + 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. @@ -126,7 +186,7 @@ initialise (the AdminUI form drops them on the next save). Pinned by ## Capability surface -`OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource` +`OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, IDriverConnectionLossSource` (`src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs:31`). | Capability | Path | Notes | @@ -253,6 +313,14 @@ the reconnect handler, #670). They are not part of `DriverHealth` or the - **Integration fixture** — exercises the driver against a reference OPC UA server (opc-plc) on the shared docker host; see [OpcUaClient-Test-Fixture.md](OpcUaClient-Test-Fixture.md) for the coverage map. +- **Wire tests against the in-process server** — `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/` + `OpcUaClientDriver*WireTests` (no Docker). `OpcUaClientDriverStallWireTests` (#676) covers the in-place blip that + used to be live-gate-only: the driver connects through `TcpStallProxy`, an in-test TCP forwarder whose pause holds + every byte without closing anything (which only works because of the endpoint rewrite above); it asserts one Bad, + a frozen read stamp, recovery on the SAME session, the identical value republished Good, a change published right + after recovery still arriving, and no source-time regression once the SDK's late backlog flush has arrived. + `OpcUaClientDriverReconnectWireTests` covers the server restart (incl. the `BadServerHalted` → `ConnectionLost` signal) + and the subscription-lost case. ## Further reading diff --git a/docs/v2/driver-specs.md b/docs/v2/driver-specs.md index 6c118870..2501b1a8 100644 --- a/docs/v2/driver-specs.md +++ b/docs/v2/driver-specs.md @@ -815,7 +815,6 @@ Already used by Client.Shared in the existing codebase. | `Password` | string? | — | For Username auth | | `SessionTimeoutMs` | int | 120000 | Server-negotiated session timeout | | `KeepAliveIntervalMs` | int | 5000 | Session keep-alive interval | -| `ReconnectPeriodMs` | int | 5000 | Initial reconnect delay | | `BrowseStrategy` | enum | Full | `Full` (browse all at startup), `Lazy`, `Hybrid` | | `BrowseRoot` | string? | — | Optional root NodeId to mirror (default: ObjectsFolder) | | `AutoAcceptCertificates` | bool | false | Accept untrusted server certificates | diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverConnectionLossSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverConnectionLossSource.cs new file mode 100644 index 00000000..29791a68 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverConnectionLossSource.cs @@ -0,0 +1,31 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +/// +/// Optional driver capability (Gitea #676) — a driver that can tell, from its protocol, that its connection is +/// DEFINITIVELY gone (it can never recover in place, only a re-initialise brings it back) raises +/// so the host re-initialises it at once instead of waiting out the #527 comms-loss +/// watchdog (~90 s of frozen ). +/// +/// +/// +/// Definitive only. An AMBIGUOUS failure — a timeout, a stalled socket, "no communication" — must NOT +/// raise: the connection may yet resume in place (a docker pause, a congested link), and tearing it +/// down would cost the session, its subscriptions and the in-place recovery. Ambiguous failures keep going +/// through the watchdog, whose conjunction (non-Healthy AND a frozen read stamp) is the honest test for them. +/// +/// +/// The host (DriverInstanceActor) acts on a raise only while the driver is Connected — in any other +/// state a re-initialise is already pending or running — and it subscribes for its own lifetime, not per +/// connect, because the instance can outlive the actor. A driver should raise at most +/// once per connection and re-arm when it is re-initialised. +/// +/// +public interface IDriverConnectionLossSource +{ + /// Raised (on the driver's own thread) when the driver's connection is definitively lost. + event EventHandler? ConnectionLost; +} + +/// Event payload for . +/// Driver-supplied reason for the diagnostic log (e.g. the protocol status that proved the loss). +public sealed record DriverConnectionLossEventArgs(string Reason); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs index b4bb7902..503adf5d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs @@ -101,12 +101,6 @@ public sealed record OpcUaClientDriverOptions /// Client-side keep-alive interval. public TimeSpan KeepAliveInterval { get; init; } = TimeSpan.FromSeconds(5); - /// - /// Retained for config compatibility only — no longer used (Gitea #670). The driver has no in-driver reconnect - /// handler; a lost session is recovered by the host's comms-loss watchdog re-initialising the driver. - /// - public TimeSpan ReconnectPeriod { get; init; } = TimeSpan.FromSeconds(5); - /// /// When true, the driver accepts any self-signed / untrusted server certificate. /// Dev-only — must be false in production so MITM attacks against the opc.tcp 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 fe76821e..a5d82caa 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -29,7 +29,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; /// back on its own recovers in place. /// /// -public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable +public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, ISkippedTagReporter, IDriverConnectionLossSource, IDisposable, IAsyncDisposable { /// Authored tags whose TagConfig carried no usable nodeId, so there is no upstream node /// to dial (Gitea #541). This driver's skip path was the quietest of the twelve: NamespaceMap @@ -83,6 +83,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit _logger = logger ?? NullLogger.Instance; _secretResolver = secretResolver ?? NullSecretResolver.Instance; _sdkTelemetry = logger is null ? SdkTelemetry.Context : SdkTelemetry.ForLogger(logger); + PostRecoveryReconcileDelay = () => _options.Timeout + _options.KeepAliveInterval; } /// Mutable so can adopt a re-parsed config on reinitialize @@ -135,6 +136,14 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// Occurs when the host connectivity status of the remote OPC UA server changes. public event EventHandler? OnHostStatusChanged; + /// + /// Gitea #676 — raised at most once per session, and only on a DEFINITIVE loss + /// (, 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 // StatusCodes are passed through verbatim per driver-specs.md §8 "cascading quality" — // downstream clients need to distinguish 'remote source down' from 'local driver failure'. @@ -164,6 +173,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// Test-only seam (#670): the most recent in-place recovery task. internal Task RecoveryTaskForTest => _recoveryTask; + /// Test seam (#676): the trailing reconcile read scheduled by the last in-place recovery. + internal Task ReconcileTaskForTest => _reconcileTask; + /// Test-only seam (#670 review M1): raise the session's PublishError as the SDK publish loop does. /// The publish error status. /// The subscription id the error names (0 = none). @@ -282,6 +294,25 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// The last in-place recovery started (#670). Test seam: lets a test await it deterministically. private Task _recoveryTask = Task.CompletedTask; + /// #676 — 1 once has been raised for the CURRENT session (once per session); + /// cleared when the session is released, so a re-initialised driver reports its next loss. + private int _connectionLostRaised; + + /// #676 — the trailing reconcile read scheduled by the last in-place recovery. Test seam. + private Task _reconcileTask = Task.CompletedTask; + + /// #676 — late notifications the replay guard dropped since it was armed, and the refs they were for; + /// summarised in ONE Information line when the window disarms. Guarded by . + private int _replayDrops; + private readonly HashSet _replayDropRefs = new(StringComparer.Ordinal); + + /// + /// Test seam (#676): how long after an in-place recovery the trailing reconcile read runs. Production: the + /// operation timeout plus one keep-alive interval — by then every request that was in flight across the stall + /// (including the Publish responses whose notifications the SDK can lose) has completed or timed out. + /// + internal Func PostRecoveryReconcileDelay { get; set; } + /// Test seam (#670 review M1): whether a subscription has stopped publishing. Production asks the SDK. internal Func IsPublishingStopped { get; set; } = static s => s.Created && s.PublishingStopped; @@ -326,6 +357,37 @@ 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. It also covers the comms-loss Bad fan + /// (). One driver-wide lock rather than one per ref: it also serialises delivery + /// ACROSS subscriptions — with several subscriptions the SDK dispatches each subscription's messages on its own + /// worker, so their notifications no longer run in parallel. That is the cost, and it is small: each critical + /// section is a dictionary update plus a subscriber call that only Tells an actor (tens of nanoseconds to + /// a few microseconds), nothing awaits under it, and per-subscription message order is unaffected. Because + /// subscribers are invoked under it, a subscriber must never block on, or call back into, this driver. + /// + private readonly Lock _deliveryLock = new(); + + /// + /// Re-review of 30a9d5f0 — per RawPath, a counter bumped (under ) every time a value is + /// delivered for it. The trailing re-read captures it before issuing its read and skips any ref delivered to since: + /// without source timestamps the read's value cannot be ordered against a notification that landed while the read + /// was in flight, and publishing it would regress the node to the value the read sampled. + /// + private readonly System.Collections.Concurrent.ConcurrentDictionary _deliverySeq = + new(StringComparer.Ordinal); + + /// 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); @@ -883,9 +945,40 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit $"No endpoint at '{endpointUrl}' matches SecurityPolicy={policy} + SecurityMode={mode}. " + $"Server advertises: {advertised}"); } + // #676: connect where the operator pointed us, not where the server says it lives. + match.EndpointUrl = RewriteEndpointUrl(match.EndpointUrl, endpointUrl); return match; } + /// + /// 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 (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. + /// + /// The EndpointUrl the server returned from GetEndpoints. + /// The URL the driver was configured with (the one GetEndpoints was sent to). + /// The URL to connect the session to. + internal static string RewriteEndpointUrl(string? advertisedUrl, string configuredUrl) + { + if (!Uri.TryCreate(configuredUrl, UriKind.Absolute, out var configured)) return advertisedUrl ?? configuredUrl; + if (string.IsNullOrWhiteSpace(advertisedUrl) + || !Uri.TryCreate(advertisedUrl, UriKind.Absolute, out var advertised)) + return configuredUrl; + if (!string.Equals(advertised.Scheme, configured.Scheme, StringComparison.OrdinalIgnoreCase)) return advertisedUrl; + // 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(); + } + /// /// Build a carrying a client user-authentication /// certificate loaded from . @@ -986,6 +1079,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit Interlocked.Exchange(ref _sessionLost, 0); Interlocked.Exchange(ref _subscriptionLost, 0); 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() + { + // Cancelled but deliberately NOT disposed (re-review of 30a9d5f0): the keep-alive thread may be reading the old + // source's Token concurrently, and a disposed source throws ObjectDisposedException there. Nothing here is + // timer-backed (no CancelAfter), so an undisposed source holds no unmanaged resource. + var old = Interlocked.Exchange(ref _sessionWorkCts, new CancellationTokenSource()); + old.Cancel(); } /// Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream. @@ -1021,8 +1128,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit // previous connection's value bookkeeping applies to them. _lastValues.Clear(); _lastGoodSourceTicks.Clear(); - _replayGuard.Clear(); - Interlocked.Exchange(ref _replayGuardUntilTicks, 0); + lock (_replayGuardLock) + { + _replayGuard.Clear(); + Interlocked.Exchange(ref _replayGuardUntilTicks, 0); + _replayDrops = 0; + _replayDropRefs.Clear(); + } _awaitingRecovery.Clear(); // #619 re-review: the last-told alarm states (and ack targets) deliberately SURVIVE a shutdown. A // ReinitializeAsync (comms-loss watchdog / Force Reconnect) is followed by the host re-subscribing the @@ -1762,20 +1874,33 @@ 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); - _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; + // 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)) + { + _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; + } + RecordUpstreamGood(fullRef, snapshot); + _lastValues[fullRef] = snapshot; + _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); + _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 . @@ -1798,14 +1923,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit if (until == 0) return false; if (Clock.GetUtcNow().UtcDateTime.Ticks >= until) { - lock (_replayGuardLock) - { - if (Interlocked.Read(ref _replayGuardUntilTicks) == until) - { - Interlocked.Exchange(ref _replayGuardUntilTicks, 0); - _replayGuard.Clear(); - } - } + DisarmReplayGuard(until); return false; } if (source is not { } src) return false; @@ -1821,11 +1939,40 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } } + /// + /// Expire the replay guard armed until — a no-op when it was re-armed or already disarmed + /// since — and log ONE Information summary of what it dropped (#676). The guard's OTel counter and per-drop Debug + /// line are invisible on a default-configured node, so before this a recovery's drops left no trace at all. + /// + /// The expiry the caller observed; only that arming is disarmed. + private void DisarmReplayGuard(long until) + { + int drops, refs; + lock (_replayGuardLock) + { + if (Interlocked.Read(ref _replayGuardUntilTicks) != until) return; + Interlocked.Exchange(ref _replayGuardUntilTicks, 0); + _replayGuard.Clear(); + drops = _replayDrops; + refs = _replayDropRefs.Count; + _replayDrops = 0; + _replayDropRefs.Clear(); + } + 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, /// for max(30 s, 2 × the longest tracked subscription keep-alive window). /// - private void ArmReplayGuard(DateTime now) + /// The window's expiry in clock ticks. + private long ArmReplayGuard(DateTime now) { var window = ReplayGuardMinimum; foreach (var sub in _subscriptions.Values.Select(r => r.Subscription)) @@ -1833,12 +1980,39 @@ 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; + } + + /// + /// #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, CancellationToken ct) + { + try + { + var wait = new DateTime(until, DateTimeKind.Utc) - Clock.GetUtcNow().UtcDateTime; + 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); + } } /// @@ -2683,7 +2857,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// session-recreate path cloned subscriptions the driver never re-tracked, and it kept health Healthy while /// it did so — the #670 symptom. A session that does not come back keeps health Degraded with the read stamp /// frozen, which is exactly what the host's #527 comms-loss watchdog trips on: ShutdownAsync + InitializeAsync - /// + re-subscribe, whose initial values come back Good. + /// + re-subscribe, whose initial values come back Good. When the keep-alive status proves the session is gone + /// for good () the driver also raises (#676), + /// so the host runs that sequence at once instead of after the watchdog's silence window. /// private void OnKeepAlive(ISession sender, KeepAliveEventArgs e) { @@ -2706,18 +2882,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit UpdateHealth(_ => new DriverHealth(DriverState.Healthy, now, null)); // A fresh grace: every subscription reports stopped right after an outage until its next message. Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0); - ArmReplayGuard(now); + var guardUntil = ArmReplayGuard(now); _diag.Record(OpcUaClientDiagnostics.KeepAliveRecovered); _logger.LogInformation( "OpcUaClientDriver '{DriverInstanceId}': upstream keep-alive recovered on the same session; " + "re-reading the refs published Bad during the outage.", _driverInstanceId); - _recoveryTask = Task.Run(() => RecoverAfterOutageAsync(sender)); + var recovery = Task.Run(() => RecoverAfterOutageAsync(sender)); + _recoveryTask = recovery; + 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; } @@ -2729,14 +2910,56 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit TransitionTo(HostState.Stopped); _diag.Record(OpcUaClientDiagnostics.KeepAliveMiss); UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"keep-alive: {e.Status}")); - if (Interlocked.Exchange(ref _sessionLost, 1) == 1) return; - if (Volatile.Read(ref _subscriptionLost) == 1) return; // already published Bad for this connection + // #676: the first bad keep-alive is often ambiguous and a definitive one follows, so this is checked on EVERY + // bad keep-alive — but only after the outage below has been published, so the Bad reaches subscribers first. + var keepAliveCode = e.Status?.StatusCode ?? StatusCodes.Good; + var definitive = IsDefinitiveSessionLoss(keepAliveCode); + // A subscription loss already published Bad for this connection. + if (Interlocked.Exchange(ref _sessionLost, 1) == 0 && Volatile.Read(ref _subscriptionLost) == 0) + { + _logger.LogWarning( + "OpcUaClientDriver '{DriverInstanceId}': upstream session keep-alive failed ({Status}); subscribed values " + + "are BadNoCommunication until it recovers or the host's comms-loss watchdog reconnects the driver.", + _driverInstanceId, e.Status); + PublishCommsLoss(now); + } + if (definitive) + RaiseConnectionLost($"upstream session keep-alive reported {StatusCodes.GetBrowseName(keepAliveCode.Code)}"); + } + /// + /// #676 — keep-alive statuses that prove the session can NEVER recover in place, so waiting out the #527 + /// watchdog would only delay the inevitable re-initialise. The server forgot the session + /// (BadSessionIdInvalid — what a re-opened channel to a restarted server answers), closed it + /// (BadSessionClosed) or never activated it (BadSessionNotActivated); or the server HALTED + /// (BadServerHalted) — measured on the #676 restart wire test, 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, so a halted session is dead for good. Everything else — BadNoCommunication, + /// BadConnectionClosed, BadNotConnected, request timeouts — is AMBIGUOUS (a stalled link that + /// resumes keeps its session, see the stall wire test) and stays with the watchdog. + /// + /// A keep-alive status. + /// True when the session is definitively gone. + internal static bool IsDefinitiveSessionLoss(StatusCode status) => + status == StatusCodes.BadSessionIdInvalid + || status == StatusCodes.BadSessionClosed + || status == StatusCodes.BadSessionNotActivated + || status == StatusCodes.BadServerHalted; + + /// #676 — raise once for the current session, isolating a faulting subscriber. + /// Why the connection is definitively gone, for the host's log. + private void RaiseConnectionLost(string reason) + { + if (Interlocked.Exchange(ref _connectionLostRaised, 1) == 1) return; _logger.LogWarning( - "OpcUaClientDriver '{DriverInstanceId}': upstream session keep-alive failed ({Status}); subscribed values " + - "are BadNoCommunication until it recovers or the host's comms-loss watchdog reconnects the driver.", - _driverInstanceId, e.Status); - PublishCommsLoss(now); + "OpcUaClientDriver '{DriverInstanceId}': connection definitively lost ({Reason}); asking the host to " + + "re-initialise now instead of waiting for the comms-loss watchdog.", _driverInstanceId, reason); + try { ConnectionLost?.Invoke(this, new DriverConnectionLossEventArgs(reason)); } + catch (Exception ex) + { + _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': a ConnectionLost subscriber faulted.", + _driverInstanceId); + } } /// @@ -2771,14 +2994,23 @@ 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}")); _logger.LogWarning( "OpcUaClientDriver '{DriverInstanceId}': {Reason}; subscribed values are BadNoCommunication until the " + - "host's comms-loss watchdog re-initialises the driver and re-subscribes.", _driverInstanceId, reason); + "host re-initialises the driver and re-subscribes.", _driverInstanceId, reason); if (Volatile.Read(ref _sessionLost) == 0) PublishCommsLoss(now); + // #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}"); } /// @@ -2801,7 +3033,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); } /// @@ -2811,18 +3044,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// private void PublishCommsLoss(DateTime now) { - foreach (var rs in _subscriptions.Values) + // Under the delivery lock (re-review of 30a9d5f0) so the Bad cannot interleave with a recovery publish. + lock (_deliveryLock) { - foreach (var pair in rs.ItemHandlers) + foreach (var rs in _subscriptions.Values) { - var fullRef = RefOf(pair.Item); - _lastValues.TryGetValue(fullRef, out var last); - _awaitingRecovery[fullRef] = 0; - // F1: stamped with the DETECTION time, as the host's own comms-loss fan does. The recorder historizes - // SourceTimestampUtc ?? ServerTimestampUtc, so the last Good's source time here put the Bad at the - // Good row's (TagId, SourceTimeUtc) key, where the insert-if-absent historian dropped it. - RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, new DataValueSnapshot( - last?.Value, StatusBadNoCommunication, now, now))); + foreach (var pair in rs.ItemHandlers) + { + var fullRef = RefOf(pair.Item); + _lastValues.TryGetValue(fullRef, out var last); + _awaitingRecovery[fullRef] = 0; + _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); + // F1: stamped with the DETECTION time, as the host's own comms-loss fan does. The recorder historizes + // SourceTimestampUtc ?? ServerTimestampUtc, so the last Good's source time here put the Bad at the + // Good row's (TagId, SourceTimeUtc) key, where the insert-if-absent historian dropped it. + RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, new DataValueSnapshot( + last?.Value, StatusBadNoCommunication, now, now))); + } } } AnnotateAlarmQuality(StatusBadNoCommunication, now); @@ -2856,15 +3094,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]); + } } } @@ -2880,6 +3116,108 @@ 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; + _deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1); + 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 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 (on ). + /// The session's background-work token. + private async Task ReconcileAfterRecoveryAsync(ISession session, Task recovery, TimeSpan delay, CancellationToken ct) + { + 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, 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; + // Captured BEFORE the read is issued: a ref delivered to while the read is in flight has a value at least as + // new as anything the read could have sampled, so the read must not overwrite it. + var seqAtIssue = refs.Select(r => _deliverySeq.GetValueOrDefault(r)).ToArray(); + 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; + lock (_deliveryLock) + { + if (_deliverySeq.GetValueOrDefault(refs[i]) != seqAtIssue[i]) continue; + 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) + _logger.LogInformation( + "OpcUaClientDriver '{DriverInstanceId}': trailing re-read after recovery found {Count} of {Total} refs " + + "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.", + _driverInstanceId); + } + } + + /// + /// 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) + { + 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. private void AnnotateAlarmQuality(uint quality, DateTime now) { @@ -2981,6 +3319,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); } @@ -3010,6 +3349,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.AdminUI/Components/Shared/Drivers/Forms/OpcUaClientDriverForm.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/OpcUaClientDriverForm.razor index 36c8f19e..a9097d97 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/OpcUaClientDriverForm.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/OpcUaClientDriverForm.razor @@ -53,11 +53,6 @@
Default 5 s.
-
- - -
Initial delay after session drop. Default 5 s.
-
@@ -330,7 +325,6 @@ public int TimeoutSeconds { get; set; } = 10; public int SessionTimeoutSeconds { get; set; } = 120; public int KeepAliveIntervalSeconds { get; set; } = 5; - public int ReconnectPeriodSeconds { get; set; } = 5; public int MaxDiscoveredNodes { get; set; } = 10_000; public int MaxBrowseDepth { get; set; } = 10; @@ -370,7 +364,6 @@ TimeoutSeconds = (int)r.Timeout.TotalSeconds, SessionTimeoutSeconds = (int)r.SessionTimeout.TotalSeconds, KeepAliveIntervalSeconds = (int)r.KeepAliveInterval.TotalSeconds, - ReconnectPeriodSeconds = (int)r.ReconnectPeriod.TotalSeconds, MaxDiscoveredNodes = r.MaxDiscoveredNodes, MaxBrowseDepth = r.MaxBrowseDepth, SecurityMode = r.SecurityMode, @@ -404,7 +397,6 @@ Timeout = TimeSpan.FromSeconds(TimeoutSeconds), SessionTimeout = TimeSpan.FromSeconds(SessionTimeoutSeconds), KeepAliveInterval = TimeSpan.FromSeconds(KeepAliveIntervalSeconds), - ReconnectPeriod = TimeSpan.FromSeconds(ReconnectPeriodSeconds), MaxDiscoveredNodes = MaxDiscoveredNodes, MaxBrowseDepth = MaxBrowseDepth, SecurityMode = SecurityMode, 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 0d691c0e..faadafff 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -153,6 +153,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// behaviour for the same reason as — a probe tick can land while the /// driver is between connects.
 private sealed record HostStatusRaised; + /// Self-sent when the wrapped driver raises (#676) + /// — it KNOWS its connection is gone for good. Marshals the event off the driver's thread. Acted on only in + /// Connected (it becomes a disconnect at once, instead of after the #527 watchdog's silence window); every other + /// behaviour is already reconnecting or connecting and ignores it. + private sealed record ConnectionLossRaised(DriverConnectionLossEventArgs Args); public sealed class RetryConnect { @@ -233,6 +238,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers private EventHandler? _alarmEventHandler; private EventHandler? _rediscoveryHandler; private EventHandler? _hostStatusHandler; + private EventHandler? _connectionLossHandler; /// When the driver last raised , and the reason /// it gave. Null until the first raise. Carried on every subsequent health snapshot so the AdminUI can @@ -400,6 +406,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // has no connection affinity, and a driver can observe a remote change while disconnected. AttachRediscoverySource(); AttachHostStatusSource(); + AttachConnectionLossSource(); PublishHealthSnapshot(); Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, _healthPollInterval); } @@ -427,6 +434,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // Stubbed drivers have no upstream alarm system — swallow the ack (it's fire-and-forget, no reply). Receive(_ => { /* stubbed drivers have no alarm backend */ }); Receive(_ => { /* stubbed drivers don't disconnect */ }); + Receive(IgnoreConnectionLoss); Receive(_ => { /* stubbed drivers don't reconnect */ }); Receive(StoreDesiredSubscriptions); // Stubbed drivers never enter Connected, so they never kick discovery; swallow defensively in case a @@ -468,6 +476,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers }); Receive(StoreDesiredSubscriptions); Receive(_ => { /* already connecting — no-op */ }); + Receive(IgnoreConnectionLoss); // 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 => @@ -500,15 +509,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // reconnects but never manages a read is measured from THIS connect, not from a stale earlier one. _connectedSinceUtc = DateTime.UtcNow; - Receive(msg => - { - _log.Warning("DriverInstance {Id}: disconnect observed ({Reason}); reconnecting", - _driverInstanceId, msg.Reason); - DetachSubscription(); - RecordFault(); - Become(Reconnecting); - PublishHealthSnapshot(); - }); + 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 (Reconnecting also swallows a stray DisconnectObserved). + Receive(msg => HandleDisconnectObserved($"driver reported connection lost: {msg.Args.Reason}")); Receive(_ => { _log.Info("DriverInstance {Id}: ForceReconnect requested by admin; re-entering Reconnecting", _driverInstanceId); @@ -764,6 +769,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(_ => { /* keep retrying via timer */ }); 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 => @@ -1101,6 +1110,50 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers _rediscoveryHandler = null; } + /// + /// 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 + /// once in PreStart and detached in PostStop, like : the + /// instance can outlive this actor, so a per-connect attach would leak a handler (holding + /// a dead Self) per respawn. + /// + private void AttachConnectionLossSource() + { + if (_driver is not IDriverConnectionLossSource src || _connectionLossHandler is not null) return; + var self = Self; + _connectionLossHandler = (_, e) => self.Tell(new ConnectionLossRaised(e)); + src.ConnectionLost += _connectionLossHandler; + } + + /// Symmetric teardown, called from PostStop — same leak argument as . + private void DetachConnectionLossSource() + { + if (_driver is IDriverConnectionLossSource src && _connectionLossHandler is not null) + src.ConnectionLost -= _connectionLossHandler; + _connectionLossHandler = null; + } + + /// + /// A connection loss reported while NOT Connected (#676): a (re)initialise is already pending or running, so + /// there is nothing to tear down — and acting on it could tear down the connection that attempt is building. + /// + private void IgnoreConnectionLoss(ConnectionLossRaised msg) => + _log.Debug("DriverInstance {Id}: driver reported connection lost ({Reason}) while not connected — ignored", + _driverInstanceId, msg.Args.Reason); + + /// + /// Connected → Reconnecting on an observed disconnect — from the #527 watchdog or a driver's definitive + /// (#676). + /// + private void HandleDisconnectObserved(string reason) + { + _log.Warning("DriverInstance {Id}: disconnect observed ({Reason}); reconnecting", _driverInstanceId, reason); + DetachSubscription(); + RecordFault(); + Become(Reconnecting); + 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 @@ -1497,6 +1550,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // same driver object), so a missing unsubscribe accumulates a handler per respawn holding a dead Self. DetachRediscoverySource(); DetachHostStatusSource(); + DetachConnectionLossSource(); try { _driver.ShutdownAsync(CancellationToken.None).GetAwaiter().GetResult(); } catch (Exception ex) { _log.Warning(ex, "DriverInstance {Id}: ShutdownAsync threw on PostStop", _driverInstanceId); } OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/OpcUaClientDriverBrowserTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/OpcUaClientDriverBrowserTests.cs index 086f9018..30945598 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/OpcUaClientDriverBrowserTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/OpcUaClientDriverBrowserTests.cs @@ -137,4 +137,17 @@ public sealed class OpcUaClientDriverBrowserTests sut.LastUsedUtc.ShouldBeGreaterThan(before, "AttributesAsync must refresh LastUsedUtc to satisfy the IBrowseSession contract"); } + + /// + /// #676: a config still carrying the deleted reconnectPeriod key must deserialise — the browser's own + /// validation (here: Certificate auth refused) is what fires, not a JSON error. + /// + [Fact] + public async Task OpenAsync_with_a_legacy_reconnectPeriod_key_still_parses() + { + var json = """{"EndpointUrl":"opc.tcp://127.0.0.1:1","reconnectPeriod":"00:00:05","AuthType":"Certificate"}"""; + var ex = await Should.ThrowAsync( + () => _sut.OpenAsync(json, TestContext.Current.CancellationToken)); + ex.Message.ShouldContain("Certificate"); + } } 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 new file mode 100644 index 00000000..d2db998f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs @@ -0,0 +1,662 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Moq; +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; + +/// +/// Gitea #676 — the driver half of the fast recovery path (), the +/// per-recovery replay-drop summary, and the trailing reconcile read after an in-place recovery. +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientConnectionLossSignalTests +{ + private const uint BadNoCommunication = 0x80310000u; + private const string SpeedRef = "Plant/dev/Speed"; + private const string CounterRef = "Plant/dev/Counter"; + + private sealed class ManualClock(DateTimeOffset start) : TimeProvider + { + public DateTimeOffset Now { get; set; } = start; + public override DateTimeOffset GetUtcNow() => Now; + public void Advance(TimeSpan by) => Now += by; + } + + private sealed class ListLogger : ILogger + { + public List<(LogLevel Level, string Message)> Entries { 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) + { + lock (Entries) Entries.Add((logLevel, formatter(state, exception))); + } + + public string[] Info(string contains) => At(LogLevel.Information, contains); + + public string[] At(LogLevel level, string contains) + { + lock (Entries) + return [.. Entries.Where(e => e.Level == level && e.Message.Contains(contains)).Select(e => e.Message)]; + } + } + + private static RawTagEntry Tag(string rawPath, string nodeId) => + new(rawPath, JsonSerializer.Serialize(new { nodeId }), WriteIdempotent: true); + + private static readonly IReadOnlyList Tags = + [ + Tag(SpeedRef, "ns=0;s=Speed"), + Tag(CounterRef, "ns=0;s=Counter"), + ]; + + private static Mock NewSession() + { + var mock = new Mock(MockBehavior.Loose); + mock.SetupGet(s => s.MessageContext).Returns(new ServiceMessageContext(SdkTelemetry.Context)); + mock.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable()); + return mock; + } + + private static (OpcUaClientDriver Driver, Mock Session, ManualClock Clock, ListLogger Log) NewConnectedDriver() + { + var log = new ListLogger(); + var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-676", log); + var session = NewSession(); + var clock = new ManualClock(new DateTimeOffset(2026, 10, 11, 12, 0, 0, TimeSpan.Zero)); + drv.Clock = clock; + drv.SetSessionForTest(session.Object); + drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); + drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, clock.Now.UtcDateTime, null)); + return (drv, session, clock, log); + } + + private static DataValue Good(object value, DateTime ts) => new(new Variant(value), StatusCodes.Good, ts, ts); + + /// Every read returns (one per requested node, in order) at . + private static void SetupRead(Mock session, Func at, params object[] values) => + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync((RequestHeader _, double _, TimestampsToReturn _, ReadValueIdCollection nodes, CancellationToken _) => + new ReadResponse + { + ResponseHeader = new ResponseHeader(), + Results = [.. nodes.Select((_, i) => Good(values[i], at()))], + DiagnosticInfos = [], + }); + + // ── IDriverConnectionLossSource ────────────────────────────────────────────────────────────────────────────── + + public static TheoryData DefinitiveStatuses => + [ + StatusCodes.BadSessionIdInvalid, + StatusCodes.BadSessionClosed, + StatusCodes.BadSessionNotActivated, + StatusCodes.BadServerHalted, + ]; + + public static TheoryData AmbiguousStatuses => + [ + StatusCodes.BadNoCommunication, + StatusCodes.BadConnectionClosed, + StatusCodes.BadNotConnected, + StatusCodes.BadRequestTimeout, + StatusCodes.BadTimeout, + StatusCodes.BadSecureChannelClosed, + ]; + + [Theory] + [MemberData(nameof(DefinitiveStatuses))] + public void A_definitive_keep_alive_raises_ConnectionLost_once_after_publishing_the_outage(uint status) + { + var (drv, _, _, _) = NewConnectedDriver(); + using var _ = drv; + drv.AddSubscriptionForTest([SpeedRef]); + var order = new List(); + drv.OnDataChange += (_, e) => order.Add($"bad:{e.Snapshot.StatusCode:X8}"); + var raised = new List(); + drv.ConnectionLost += (_, e) => { raised.Add(e); order.Add("lost"); }; + + drv.RaiseKeepAliveForTest(status); + drv.RaiseKeepAliveForTest(status); + drv.RaiseKeepAliveForTest(StatusCodes.BadSessionIdInvalid); + + raised.ShouldHaveSingleItem("once per session, however many definitive keep-alives follow") + .Reason.ShouldContain(StatusCodes.GetBrowseName(status)); + order.ShouldBe([$"bad:{BadNoCommunication:X8}", "lost"], "subscribers see the Bad before the host tears down"); + } + + [Theory] + [MemberData(nameof(AmbiguousStatuses))] + public void An_ambiguous_keep_alive_does_not_raise_ConnectionLost(uint status) + { + var (drv, _, _, _) = NewConnectedDriver(); + using var _ = drv; + drv.AddSubscriptionForTest([SpeedRef]); + var raised = 0; + drv.ConnectionLost += (_, _) => raised++; + var seen = new List(); + drv.OnDataChange += (_, e) => seen.Add(e); + + for (var i = 0; i < 5; i++) drv.RaiseKeepAliveForTest(status); + + // Positive control: the keep-alives WERE processed as an outage — the outage just is not definitive. + seen.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(BadNoCommunication); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + raised.ShouldBe(0, "an ambiguous loss may yet resume in place — it belongs to the #527 watchdog"); + } + + [Fact] + public void An_ambiguous_outage_that_turns_definitive_raises_once() + { + var (drv, _, _, _) = NewConnectedDriver(); + using var _ = drv; + var raised = 0; + drv.ConnectionLost += (_, _) => raised++; + + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + raised.ShouldBe(0); + drv.RaiseKeepAliveForTest(StatusCodes.BadServerHalted); + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.BadServerHalted); + + raised.ShouldBe(1); + } + + [Fact] + public async Task The_signal_is_re_armed_by_a_re_initialise() + { + var (drv, _, _, _) = NewConnectedDriver(); + using var _ = drv; + var raised = 0; + drv.ConnectionLost += (_, _) => raised++; + drv.RaiseKeepAliveForTest(StatusCodes.BadSessionIdInvalid); + raised.ShouldBe(1); + + // The host's response: ShutdownAsync + InitializeAsync, i.e. a NEW session. + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + var fresh = NewSession(); + drv.SetSessionForTest(fresh.Object); + drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); + drv.RaiseKeepAliveForTest(StatusCodes.BadSessionIdInvalid); + + raised.ShouldBe(2, "the next session's definitive loss must be reported too"); + } + + [Fact] + public void A_definitive_keep_alive_from_a_superseded_session_is_ignored() + { + var (drv, _, _, _) = NewConnectedDriver(); + using var _ = drv; + var raised = 0; + drv.ConnectionLost += (_, _) => raised++; + + drv.RaiseKeepAliveForTest(StatusCodes.BadSessionIdInvalid, NewSession().Object); + + raised.ShouldBe(0); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + [Fact] + public void A_subscription_the_server_dropped_raises_ConnectionLost_once() + { + var (drv, _, _, _) = NewConnectedDriver(); + using var _ = drv; + drv.AddSubscriptionForTest([SpeedRef]); + var raised = new List(); + drv.ConnectionLost += (_, e) => raised.Add(e); + + drv.RaisePublishErrorForTest(StatusCodes.BadSubscriptionIdInvalid, subscriptionId: 4242); // not ours + raised.ShouldBeEmpty(); + + drv.RaisePublishErrorForTest(StatusCodes.BadNoSubscription); + drv.RaisePublishErrorForTest(StatusCodes.BadNoSubscription); + + raised.ShouldHaveSingleItem().Reason.ShouldContain("subscription lost"); + } + + [Fact] + public void A_faulting_ConnectionLost_subscriber_does_not_break_the_keep_alive_path() + { + var (drv, _, _, _) = NewConnectedDriver(); + using var _ = drv; + drv.ConnectionLost += (_, _) => throw new InvalidOperationException("subscriber bug"); + + Should.NotThrow(() => drv.RaiseKeepAliveForTest(StatusCodes.BadSessionClosed)); + 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] + public async Task The_replay_window_logs_one_Information_summary_of_its_drops_when_it_closes() + { + var (drv, session, clock, log) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.FromDays(1); // keep the trailing read out of this test + var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0.AddSeconds(5))); + drv.DeliverNotificationForTest(handle, CounterRef, Good(2, t0.AddSeconds(5))); + SetupRead(session, () => t0.AddSeconds(5), 1, 2); + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + + // Three late replays across two refs, one fresh notification that is accepted. + drv.DeliverNotificationForTest(handle, SpeedRef, Good(0, t0)); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0.AddSeconds(5))); + drv.DeliverNotificationForTest(handle, CounterRef, Good(0, t0)); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(3, t0.AddSeconds(6))); + log.Info("replay window closed").ShouldBeEmpty("the summary is logged once, when the window closes"); + + clock.Advance(TimeSpan.FromSeconds(31)); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(4, t0.AddSeconds(40))); // first notification after expiry + drv.DeliverNotificationForTest(handle, SpeedRef, Good(5, t0.AddSeconds(41))); + + log.Info("replay window closed").ShouldHaveSingleItem() + .ShouldContain("dropped 3 late notifications across 2 refs after recovery"); + } + + [Fact] + 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; + 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; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(0, t0.AddSeconds(-1))); // one drop in window 1 + + 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."], + "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 ────────────────────────────────────────────────────────────────────────────────── + + [Fact] + public async Task The_trailing_re_read_publishes_a_change_the_subscription_lost_and_nothing_it_already_has() + { + var (drv, session, clock, log) = 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)); + + // The recovery read sees both unchanged; by the trailing read Speed has moved on (its notification was lost in + // the stalled publish pipeline) while Counter has not. + var reads = 0; + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(() => + { + var trailing = Interlocked.Increment(ref reads) > 1; + return new ReadResponse + { + ResponseHeader = new ResponseHeader(), + Results = trailing + ? [Good(40f, t0.AddSeconds(9)), Good(7, t0)] + : [Good(10f, t0), Good(7, t0)], + DiagnosticInfos = [], + }; + }); + var seen = new List(); + drv.OnDataChange += (_, e) => { lock (seen) seen.Add(e); }; + + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + await drv.ReconcileTaskForTest; + + reads.ShouldBe(2, "the recovery read plus exactly one trailing read"); + DataChangeEventArgs[] afterRecovery; + lock (seen) afterRecovery = [.. seen.Where(e => e.Snapshot.StatusCode == 0u)]; + // Recovery read republishes both Good (they were published Bad); the trailing read adds ONLY the lost change. + afterRecovery.Select(e => (e.FullReference, e.Snapshot.Value)) + .ShouldBe([(SpeedRef, (object?)10f), (CounterRef, (object?)7), (SpeedRef, (object?)40f)]); + log.Info("trailing re-read").ShouldHaveSingleItem().ShouldContain("found 1 of 2 refs"); + } + + [Fact] + public async Task The_trailing_re_read_never_publishes_an_older_value_than_a_notification_already_accepted() + { + var (drv, session, clock, _) = NewConnectedDriver(); + using var _ = drv; + var gate = new TaskCompletionSource(); + drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero; + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0)); + var reads = 0; + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .Returns(async () => + { + if (Interlocked.Increment(ref reads) > 1) await gate.Task; // hold the trailing read + return new ReadResponse + { + ResponseHeader = new ResponseHeader(), + Results = [Good(reads > 1 ? 20f : 10f, t0.AddSeconds(reads > 1 ? 5 : 0))], + DiagnosticInfos = [], + }; + }); + var seen = new List<(uint Status, object? Value)>(); + drv.OnDataChange += (_, e) => { lock (seen) seen.Add((e.Snapshot.StatusCode, e.Snapshot.Value)); }; + + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + // While the trailing read is in flight, a NEWER value arrives through the subscription. + drv.DeliverNotificationForTest(handle, SpeedRef, Good(30f, t0.AddSeconds(8))); + gate.SetResult(); + await drv.ReconcileTaskForTest; + + reads.ShouldBe(2, "the trailing read must actually have run, or the absence below proves nothing"); + lock (seen) + seen.ShouldBe([(BadNoCommunication, (object?)10f), (0u, 10f), (0u, 30f)], + "the trailing read's 20 @ t0+5 is older than the 30 @ t0+8 already accepted, so it is not published"); + } + + [Fact] + public async Task The_trailing_re_read_does_nothing_if_the_session_was_replaced() + { + var (drv, session, clock, _) = NewConnectedDriver(); + using var _ = drv; + drv.PostRecoveryReconcileDelay = () => TimeSpan.FromMilliseconds(300); + var handle = drv.AddSubscriptionForTest([SpeedRef]); + var t0 = clock.Now.UtcDateTime; + drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0)); + SetupRead(session, () => t0.AddSeconds(5), 99f); + + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + // A re-initialise lands between the recovery read and the trailing one. + drv.SetSessionForTest(NewSession().Object); + await drv.ReconcileTaskForTest; + + session.Verify(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + 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); + } + + /// + /// Re-review of 30a9d5f0: with no source timestamps to compare, "newer" cannot be judged from the values. The + /// trailing read samples V1; while it is in flight a live notification delivers V2; the read then returns V1 — + /// different from the last delivered value and with no timestamp to prove it older. It must NOT be published: a + /// ref delivered to after the read was issued is skipped (per-ref delivery sequence), or the node regresses to V1. + /// + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_notification_delivered_while_the_trailing_read_is_in_flight_is_not_overwritten(bool readHasSourceTime) + { + 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 = 0; + session.Setup(s => s.ReadAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(() => + { + DataValue result; + if (Interlocked.Increment(ref reads) == 1) + result = NoSource(10f, t0.AddSeconds(1)); // recovery read: unchanged + else + { + // The trailing read has sampled V1 at the server; before its response is processed, the live + // subscription delivers the newer V2. + drv.DeliverNotificationForTest(handle, SpeedRef, NoSource(20f, t0.AddSeconds(9))); + result = readHasSourceTime + ? Good(15f, t0.AddSeconds(8)) // read carries a source time, V2 did not + : NoSource(15f, t0.AddSeconds(8)); + } + return new ReadResponse { ResponseHeader = new ResponseHeader(), Results = [result], DiagnosticInfos = [] }; + }); + + var seen = await RecoverAndReconcileAsync(drv); + + reads.ShouldBe(2, "the trailing read must have run"); + lock (seen) seen.Select(e => e.Value).ShouldBe([10f, 10f, 20f], "the read's stale 15 must not overwrite the newer 20"); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverFactoryTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverFactoryTests.cs index 1c6d2571..b0ac2c8a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverFactoryTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverFactoryTests.cs @@ -75,4 +75,23 @@ public class OpcUaClientDriverFactoryTests opts.RawTags[0].WriteIdempotent.ShouldBeFalse(); opts.RawTags[1].WriteIdempotent.ShouldBeTrue(); } + + /// + /// #676 deleted the dead ReconnectPeriod option (unused since #670 removed the SDK reconnect handler). + /// DriverConfig rows stored before that still carry it and must keep parsing — and the keys around it must + /// still bind, so the unknown member is skipped rather than aborting the read. + /// + [Fact] + public void A_stored_config_still_carrying_reconnectPeriod_parses() + { + const string legacy = + """{"endpointUrl":"opc.tcp://host:4840","reconnectPeriod":"00:00:05","keepAliveInterval":"00:00:07"}"""; + + var opts = OpcUaClientDriverFactoryExtensions.ParseOptions("drv-legacy", legacy); + opts.EndpointUrl.ShouldBe("opc.tcp://host:4840"); + opts.KeepAliveInterval.ShouldBe(TimeSpan.FromSeconds(7)); + + var driver = OpcUaClientDriverFactoryExtensions.CreateInstance("drv-legacy", legacy, NullLoggerFactory.Instance); + driver.DriverInstanceId.ShouldBe("drv-legacy"); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverProbeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverProbeTests.cs index 5ebb4e0e..971c85b7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverProbeTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverProbeTests.cs @@ -134,4 +134,27 @@ public sealed class OpcUaClientDriverProbeTests await acceptTask.WaitAsync(TimeSpan.FromSeconds(5), testCt); } } + + /// + /// #676: a stored config still carrying the deleted reconnectPeriod key must reach the connect phase, not + /// fail as invalid JSON — the probe parses with the same skip-unknown-members options as the factory. + /// + [Fact] + public async Task A_config_still_carrying_reconnectPeriod_is_probed_not_rejected() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + + var result = await _probe.ProbeAsync( + $$"""{"EndpointUrl":"opc.tcp://127.0.0.1:{{port}}","reconnectPeriod":"00:00:05"}""", + TimeSpan.FromSeconds(5), + TestContext.Current.CancellationToken); + + result.Ok.ShouldBeFalse(); + var message = result.Message.ShouldNotBeNull(); + message.ShouldNotContain("invalid", Case.Insensitive); + message.ShouldContain("Connect failed", Case.Insensitive); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverScaffoldTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverScaffoldTests.cs index 78cc3d26..a15b4b6d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverScaffoldTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientDriverScaffoldTests.cs @@ -31,7 +31,6 @@ public sealed class OpcUaClientDriverScaffoldTests var opts = new OpcUaClientDriverOptions(); opts.SessionTimeout.ShouldBe(TimeSpan.FromSeconds(120)); opts.KeepAliveInterval.ShouldBe(TimeSpan.FromSeconds(5)); - opts.ReconnectPeriod.ShouldBe(TimeSpan.FromSeconds(5)); } /// Verifies that driver reports type and ID before connecting. 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 new file mode 100644 index 00000000..6e9ce0e8 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientEndpointRewriteTests.cs @@ -0,0 +1,69 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; + +/// +/// Gitea #676 — the driver connects to the endpoint URL the server ADVERTISES in GetEndpoints. A server behind NAT, +/// a port map or a proxy advertises an address the client cannot reach (or one that bypasses the proxy), so the +/// driver must rewrite host AND port to the configured URL when the scheme matches — what the SDK's own +/// CoreClientUtils.SelectEndpoint does. +/// +public sealed class OpcUaClientEndpointRewriteTests +{ + [Fact] + public void Host_and_port_of_the_advertised_url_are_replaced_by_the_configured_ones() + { + OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://internal-host:4840/OtOpcUa", "opc.tcp://10.1.2.3:54840/OtOpcUa") + .ShouldBe("opc.tcp://10.1.2.3:54840/OtOpcUa"); + } + + [Fact] + public void The_advertised_path_is_kept() + { + // The path is the server's to choose (several endpoints may share a host:port); only the address is ours. + OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://plc-a:4840/Server/Ua", "opc.tcp://gateway:6000/") + .ShouldBe("opc.tcp://gateway:6000/Server/Ua"); + } + + [Fact] + public void Only_the_port_differs_still_rewrites() + { + // The in-process stall proxy shape: same host, the configured port is the proxy's. + OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://127.0.0.1:4840/OtOpcUa", "opc.tcp://127.0.0.1:61000/OtOpcUa") + .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() + { + // An https endpoint advertised next to an opc.tcp one is a different transport, not a NAT artefact. + OpcUaClientDriver.RewriteEndpointUrl("https://srv:443/ua", "opc.tcp://srv:4840/ua") + .ShouldBe("https://srv:443/ua"); + } + + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData("not a url")] + public void An_unusable_advertised_url_falls_back_to_the_configured_one(string? advertised) + { + OpcUaClientDriver.RewriteEndpointUrl(advertised, "opc.tcp://srv:4840/ua").ShouldBe("opc.tcp://srv:4840/ua"); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReconnectTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReconnectTests.cs index 88114079..0f1d1901 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReconnectTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReconnectTests.cs @@ -4,29 +4,15 @@ using Xunit; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; /// -/// Scaffold tests for the reconnect options. Since Gitea #670 the driver has no in-driver reconnect -/// handler (session loss is reported and the host's comms-loss watchdog re-initialises the driver); -/// ReconnectPeriod is retained for config compatibility only. Outage behaviour is covered by -/// and, over the wire, OpcUaClientDriverReconnectWireTests. +/// Reconnect scaffolding. Since Gitea #670 the driver has no in-driver reconnect handler (session loss is reported +/// and the host re-initialises the driver — at once on a definitive loss, #676, else via the comms-loss watchdog), +/// and #676 deleted the dead ReconnectPeriod option; pins that +/// stored configs still carrying it parse. Outage behaviour is covered by , +/// and, over the wire, OpcUaClientDriverReconnectWireTests. /// [Trait("Category", "Unit")] public sealed class OpcUaClientReconnectTests { - /// Verifies that the default reconnect period matches the driver specification of 5 seconds. - [Fact] - public void Default_ReconnectPeriod_matches_driver_specs_5_seconds() - { - new OpcUaClientDriverOptions().ReconnectPeriod.ShouldBe(TimeSpan.FromSeconds(5)); - } - - /// Verifies that reconnect period can be configured for aggressive or relaxed retries. - [Fact] - public void Options_ReconnectPeriod_is_configurable_for_aggressive_or_relaxed_retry() - { - var opts = new OpcUaClientDriverOptions { ReconnectPeriod = TimeSpan.FromMilliseconds(500) }; - opts.ReconnectPeriod.ShouldBe(TimeSpan.FromMilliseconds(500)); - } - /// Verifies that the driver starts in the Unknown state before initialization. [Fact] public void Driver_starts_unknown_pre_init() diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/OpcUaClientDriverPageFormSerializationTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/OpcUaClientDriverPageFormSerializationTests.cs index f0002a9b..eede6ea9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/OpcUaClientDriverPageFormSerializationTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/OpcUaClientDriverPageFormSerializationTests.cs @@ -34,7 +34,6 @@ public sealed class OpcUaClientDriverPageFormSerializationTests Timeout = TimeSpan.FromSeconds(20), SessionTimeout = TimeSpan.FromSeconds(180), KeepAliveInterval = TimeSpan.FromSeconds(10), - ReconnectPeriod = TimeSpan.FromSeconds(15), AutoAcceptCertificates = true, BrowseRoot = "i=85", MaxDiscoveredNodes = 5000, @@ -58,7 +57,6 @@ public sealed class OpcUaClientDriverPageFormSerializationTests back.Timeout.ShouldBe(TimeSpan.FromSeconds(20)); back.SessionTimeout.ShouldBe(TimeSpan.FromSeconds(180)); back.KeepAliveInterval.ShouldBe(TimeSpan.FromSeconds(10)); - back.ReconnectPeriod.ShouldBe(TimeSpan.FromSeconds(15)); back.AutoAcceptCertificates.ShouldBeTrue(); back.BrowseRoot.ShouldBe("i=85"); back.MaxDiscoveredNodes.ShouldBe(5000); @@ -105,7 +103,6 @@ public sealed class OpcUaClientDriverPageFormSerializationTests Timeout = TimeSpan.FromSeconds(12), SessionTimeout = TimeSpan.FromSeconds(150), KeepAliveInterval = TimeSpan.FromSeconds(7), - ReconnectPeriod = TimeSpan.FromSeconds(8), AutoAcceptCertificates = true, BrowseRoot = "i=85", MaxDiscoveredNodes = 3000, @@ -133,7 +130,6 @@ public sealed class OpcUaClientDriverPageFormSerializationTests result.Timeout.ShouldBe(TimeSpan.FromSeconds(12)); result.SessionTimeout.ShouldBe(TimeSpan.FromSeconds(150)); result.KeepAliveInterval.ShouldBe(TimeSpan.FromSeconds(7)); - result.ReconnectPeriod.ShouldBe(TimeSpan.FromSeconds(8)); result.AutoAcceptCertificates.ShouldBeTrue(); result.BrowseRoot.ShouldBe("i=85"); result.MaxDiscoveredNodes.ShouldBe(3000); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs index 17f6f54a..d1c645c9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs @@ -13,6 +13,8 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// comms-loss watchdog trips on). Recovery is the watchdog's sequence, replayed here as DriverInstanceActor /// runs it: ShutdownAsync, InitializeAsync, re-subscribe — after which the value comes back Good. There is no /// in-driver reconnect any more (#670: the SDK reconnect handler kept everything Good and Healthy forever). +/// Since #676 a gracefully stopped server is a DEFINITIVE loss (BadServerHalted), so the driver also raises +/// ConnectionLost and the host runs that sequence at once instead of after the watchdog's silence window. /// [Trait("Category", "Integration")] public sealed class OpcUaClientDriverReconnectWireTests @@ -39,6 +41,8 @@ public sealed class OpcUaClientDriverReconnectWireTests OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f); await using var drv = OpcUaClientDriverWireTests.NewDriver(rig, "wire-reconnect"); + var lost = new List(); + drv.ConnectionLost += (_, e) => { lock (lost) lost.Add(e.Reason); }; await drv.InitializeAsync("{}", ct); var seen = new List<(uint Status, object? Value)>(); @@ -67,12 +71,20 @@ public sealed class OpcUaClientDriverReconnectWireTests frozen.ShouldNotBeNull(); await Task.Delay(TimeSpan.FromSeconds(3), ct); // three keep-alive intervals drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen); + + // #676: a server that HALTED (graceful stop) is a definitive loss — measured here, the SDK reports + // BadServerHalted on every keep-alive and never re-opens the channel, so the session can never recover in + // place. The driver says so at once, so the host re-initialises it instead of waiting out the ~90 s watchdog. + await WaitAsync(() => { lock (lost) return lost.Count > 0; }, + "a halted upstream must raise ConnectionLost", TimeSpan.FromSeconds(10)); + lock (lost) lost.Single().ShouldContain("BadServerHalted"); }); // The restarted server knows nothing of the old session, so the driver stays down on its own … await Task.Delay(TimeSpan.FromSeconds(3), ct); drv.GetHealth().State.ShouldNotBe(DriverState.Healthy, "nothing in the driver may paper over a dead session"); lock (seen) seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1, "one Bad per outage, not per keep-alive"); + lock (lost) lost.Count.ShouldBe(1, "once per session — the restarted server's continued refusals raise nothing more"); // … until the host's comms-loss watchdog re-initialises it (DriverInstanceActor: Reconnecting → ShutdownAsync + // InitializeAsync → Connected → re-subscribe). The new server is empty: redeploy the nodes first, holding the @@ -100,6 +112,7 @@ public sealed class OpcUaClientDriverReconnectWireTests var back = await drv.ReadAsync([OpcUaClientDriverWireTests.SpeedRef], ct); back[0].StatusCode.ShouldBe(0u); Convert.ToSingle(back[0].Value).ShouldBe(88.0f); + lock (lost) lost.Count.ShouldBe(1, "the healthy re-initialised session reports no loss"); } /// @@ -117,6 +130,8 @@ public sealed class OpcUaClientDriverReconnectWireTests OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 5.0f); await using var drv = OpcUaClientDriverWireTests.NewDriver(rig, "wire-sublost"); + var lost = new List(); + drv.ConnectionLost += (_, e) => { lock (lost) lost.Add(e.Reason); }; await drv.InitializeAsync("{}", ct); var seen = new List<(uint Status, object? Value)>(); drv.OnDataChange += (_, e) => @@ -140,6 +155,8 @@ public sealed class OpcUaClientDriverReconnectWireTests drv.GetHealth().State.ShouldNotBe(DriverState.Healthy, "… and must not revive a driver whose data plane is dead"); drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen); drv.GetHostStatuses().Single().State.ShouldBe(HostState.Running, "the session itself is fine"); + // #676: a subscription the server dropped never comes back on this session — a definitive loss. + lock (lost) lost.ShouldHaveSingleItem().ShouldContain("subscription lost"); // The watchdog's re-initialise recreates the subscription and the value comes back Good. await drv.ShutdownAsync(ct); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverStallWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverStallWireTests.cs new file mode 100644 index 00000000..82504d64 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverStallWireTests.cs @@ -0,0 +1,159 @@ +using Opc.Ua; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #676 item 1 — the IN-PLACE blip (#670): the upstream stops answering but the TCP connection and the +/// session survive, keep-alives fail, then traffic resumes before the session times out. Until now only the live +/// gate (docker pause of opc-plc) covered it. reproduces it on loopback: the driver +/// connects THROUGH the proxy (which works only because of the #676 endpoint rewrite — the server advertises its own +/// port, which would bypass the proxy), and pausing the proxy stalls every byte without closing anything. +/// +/// +/// Writing this test found a real defect: a change the server published right after the stall ended rode a Publish +/// response that had been in flight across the stall; the SDK timed that request out, dropped the response and never +/// republished it, so the node stayed Good on the PRE-change value indefinitely. The driver's trailing reconcile read +/// (ReconcileAfterRecoveryAsync) is the fix; the "40.0 arrives" leg below goes red without it. +/// +[Trait("Category", "Integration")] +public sealed class OpcUaClientDriverStallWireTests +{ + private const uint BadNoCommunication = 0x80310000u; + + private static async Task WaitAsync(Func condition, string because, TimeSpan? timeout = null) + { + var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(30)); + while (DateTime.UtcNow < deadline) + { + if (condition()) return; + await Task.Delay(25); + } + condition().ShouldBeTrue(because); + } + + private sealed record Seen(uint Status, object? Value, DateTime? Source); + + [Fact] + public async Task A_stalled_connection_goes_Bad_once_and_recovers_in_place_on_the_same_session_without_regressing() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-stall", ct); + 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 + // 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), + sessionTimeout: TimeSpan.FromSeconds(120), subscriptionLifetimeCount: 3000); + var connectionLost = 0; + drv.ConnectionLost += (_, _) => Interlocked.Increment(ref connectionLost); + await drv.InitializeAsync("{}", ct); + proxy.AcceptedConnections.ShouldBeGreaterThan(0, "the session must run THROUGH the proxy, or pausing it proves nothing"); + + var seen = new List(); + drv.OnDataChange += (_, e) => + { + if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef) + lock (seen) seen.Add(new Seen(e.Snapshot.StatusCode, e.Snapshot.Value, e.Snapshot.SourceTimestampUtc)); + }; + await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct); + await WaitAsync(() => { lock (seen) return seen.Any(s => s is { Status: 0u, Value: 10.0f }); }, "initial value"); + + 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)) 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. + uint maxSequence = 0; + var lateMessages = 0; + session.Notification += (_, e) => + { + var msg = e.NotificationMessage; + if (msg is null) return; + lock (seen) + { + if (msg.SequenceNumber < maxSequence && msg.NotificationData.Count > 0) lateMessages++; + maxSequence = Math.Max(maxSequence, msg.SequenceNumber); + } + }; + + // ── Stall ──────────────────────────────────────────────────────────────────────────────────────────────── + proxy.Pause(); + await WaitAsync(() => { lock (seen) return seen.Any(s => s.Status == BadNoCommunication); }, + "a stalled upstream must publish BadNoCommunication for the subscribed ref"); + lock (seen) seen.Single(s => s.Status == BadNoCommunication).Value.ShouldBe(10.0f, "the Bad retains the value"); + drv.GetHealth().State.ShouldNotBe(DriverState.Healthy); + var frozen = drv.GetHealth().LastSuccessfulRead.ShouldNotBeNull(); + + // The server's value moves on while the client cannot see it: away and back to the pre-outage value, so the + // recovered value is IDENTICAL to the one published Bad — the (value, status) dedup trap. + OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 20.0f); + 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 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"); + + // ── Resume well inside the 30 s session timeout ────────────────────────────────────────────────────────── + proxy.Resume(); + await WaitAsync(() => drv.GetHealth().State == DriverState.Healthy, "a resumed stall must recover in place"); + int badIndex; + lock (seen) badIndex = seen.FindIndex(s => s.Status == BadNoCommunication); + await WaitAsync(() => { lock (seen) return seen.Skip(badIndex + 1).Any(s => s is { Status: 0u, Value: 10.0f }); }, + "the identical value must be republished Good after the outage"); + + drv.Session.ShouldBeSameAs(session, "an in-place blip must not cost the session"); + proxy.AcceptedConnections.ShouldBe(connectionsBefore, "and not the TCP connection either"); + + // A change published straight after recovery — while the Publish requests held across the stall are still + // 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)", + 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", + TimeSpan.FromSeconds(60)); + + lock (seen) + { + var good = seen.Where(s => s.Status == 0u && s.Source is not null).ToList(); + for (var i = 1; i < good.Count; i++) + good[i].Source!.Value.ShouldBeGreaterThanOrEqualTo(good[i - 1].Source!.Value, + $"published Good value #{i} ({good[i].Value}) regressed in source time — a late replay reached the " + + $"node. Sequence: [{string.Join(", ", good.Select(s => $"{s.Value}@{s.Source:HH:mm:ss.fff}"))}]"); + seen.Last().Value.ShouldBe(40.0f, "nothing older may land after the latest value"); + seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1); + } + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + drv.Session.ShouldBeSameAs(session); + Volatile.Read(ref connectionLost).ShouldBe(0, + "a stall is AMBIGUOUS (the session survived it) — it must go through the #527 watchdog, never the fast path"); + } +} 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 f55ae238..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,10 +41,11 @@ public sealed class OpcUaClientDriverWireTests internal static OpcUaClientDriver NewDriver( ServerEventRig rig, string id, string? browseRoot = null, IReadOnlyList? extraTags = null, - TimeSpan? reconnectPeriod = null) => + string? endpointUrl = null, TimeSpan? timeout = null, TimeSpan? sessionTimeout = null, + uint? subscriptionLifetimeCount = null) => new(new OpcUaClientDriverOptions { - EndpointUrl = rig.EndpointUrl, + EndpointUrl = endpointUrl ?? rig.EndpointUrl, RawTags = [ Tag(SpeedRef, ServerEventRig.RawNsu(SpeedId)), @@ -58,11 +59,11 @@ public sealed class OpcUaClientDriverWireTests Username = "apikey", Password = ServerEventRig.KeyPassword, AutoAcceptCertificates = true, - Timeout = TimeSpan.FromSeconds(10), - SessionTimeout = TimeSpan.FromSeconds(30), + Timeout = timeout ?? TimeSpan.FromSeconds(10), + SessionTimeout = sessionTimeout ?? TimeSpan.FromSeconds(30), + SubscriptionLifetimeCount = subscriptionLifetimeCount ?? 1000, KeepAliveInterval = TimeSpan.FromSeconds(1), BrowseRoot = browseRoot, - ReconnectPeriod = reconnectPeriod ?? TimeSpan.FromSeconds(5), }, id); internal static async Task StartRigAsync(string name, CancellationToken ct) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/TcpStallProxy.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/TcpStallProxy.cs new file mode 100644 index 00000000..a808c0cf --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/TcpStallProxy.cs @@ -0,0 +1,126 @@ +using System.Net; +using System.Net.Sockets; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #676 — an in-test TCP forwarder that can STALL a connection without closing it: the shape of a +/// docker pause, a hung switch port or a congested WAN link, where the socket stays open and simply stops +/// carrying bytes. Each accepted connection gets a fresh upstream connection and two copy pumps; every pump awaits a +/// shared gate before each write, so holds all traffic in both directions (bytes already read +/// are held, the rest back up in the kernel buffers) and lets it all through in order. Nothing +/// is ever closed by a pause — the outage is invisible at the TCP level, exactly as in production. +/// +internal sealed class TcpStallProxy : IAsyncDisposable +{ + private readonly TcpListener _listener = new(IPAddress.Loopback, 0); + private readonly int _targetPort; + private readonly CancellationTokenSource _cts = new(); + private readonly List _sockets = []; + private readonly Lock _lock = new(); + private TaskCompletionSource _gate = Open(); + private Task _acceptLoop = Task.CompletedTask; + + private TcpStallProxy(int targetPort) => _targetPort = targetPort; + + /// The loopback port clients connect to. + public int Port => ((IPEndPoint)_listener.LocalEndpoint).Port; + + /// Number of client connections accepted so far — a new one means the client re-connected. + public int AcceptedConnections { get; private set; } + + /// Starts forwarding 127.0.0.1:{Port} to 127.0.0.1:{targetPort}. + public static TcpStallProxy Start(int targetPort) + { + var proxy = new TcpStallProxy(targetPort); + proxy._listener.Start(); + proxy._acceptLoop = proxy.AcceptLoopAsync(); + return proxy; + } + + /// Rewrites the port of an opc.tcp URL to this proxy's. + public string Through(string endpointUrl) => new UriBuilder(endpointUrl) { Port = Port }.Uri.ToString(); + + /// Stops all traffic in both directions; sockets stay open. + public void Pause() + { + lock (_lock) + if (_gate.Task.IsCompleted) _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + + /// Releases everything held since , in order. + public void Resume() + { + lock (_lock) _gate.TrySetResult(); + } + + private static TaskCompletionSource Open() + { + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + tcs.SetResult(); + return tcs; + } + + private Task CurrentGate() + { + lock (_lock) return _gate.Task; + } + + private async Task AcceptLoopAsync() + { + while (!_cts.IsCancellationRequested) + { + TcpClient client; + try { client = await _listener.AcceptTcpClientAsync(_cts.Token); } + catch { return; } + var upstream = new TcpClient(); + try { await upstream.ConnectAsync(IPAddress.Loopback, _targetPort, _cts.Token); } + catch + { + client.Dispose(); + upstream.Dispose(); + continue; + } + lock (_lock) + { + AcceptedConnections++; + _sockets.Add(client); + _sockets.Add(upstream); + } + _ = PumpAsync(client.GetStream(), upstream.GetStream()); + _ = PumpAsync(upstream.GetStream(), client.GetStream()); + } + } + + private async Task PumpAsync(NetworkStream from, NetworkStream to) + { + var buffer = new byte[64 * 1024]; + try + { + while (true) + { + var n = await from.ReadAsync(buffer, _cts.Token); + if (n == 0) break; + await CurrentGate().WaitAsync(_cts.Token); + await to.WriteAsync(buffer.AsMemory(0, n), _cts.Token); + } + } + catch { /* either side gone or the proxy disposed */ } + // Half-close propagation is not needed: a real close (server stopped) tears the pair down on the next read. + try { to.Socket.Shutdown(SocketShutdown.Send); } catch { } + } + + public async ValueTask DisposeAsync() + { + Resume(); + await _cts.CancelAsync(); + _listener.Stop(); + try { await _acceptLoop; } catch { } + lock (_lock) + { + foreach (var s in _sockets) s.Dispose(); + _sockets.Clear(); + } + _cts.Dispose(); + } +} 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 new file mode 100644 index 00000000..17d0c2f7 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConnectionLossSignalTests.cs @@ -0,0 +1,175 @@ +using Akka.Actor; +using Akka.Event; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Gitea #676 — a driver that KNOWS its connection is gone () gets +/// re-initialised at once instead of after the #527 watchdog's silence window (~90 s in production). The watchdog is +/// configured at an hour here, so anything that reconnects within seconds can only be the new signal. +/// +public sealed class DriverInstanceActorConnectionLossSignalTests : RuntimeActorTestBase +{ + private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(50); + + /// A permanently Healthy (fresh-read) stub — the watchdog can never fire for it — that can raise + /// on demand and counts its attached handlers. + private sealed class LossSourceStubDriver : StubDriver, IDriverConnectionLossSource + { + private EventHandler? _handlers; + private int _handlerCount; + + private int _healthReads; + + public int HandlerCount => Volatile.Read(ref _handlerCount); + + /// Health-poll observations — the positive control for an absence assertion. + public int HealthReadCount => Volatile.Read(ref _healthReads); + + public override DriverHealth GetHealth() + { + Interlocked.Increment(ref _healthReads); + return base.GetHealth(); + } + + public event EventHandler? ConnectionLost + { + add { _handlers += value; Interlocked.Increment(ref _handlerCount); } + remove { _handlers -= value; Interlocked.Decrement(ref _handlerCount); } + } + + /// Raised from a pool thread, as a real driver's keep-alive callback would. + public Task RaiseAsync(string reason) => + Task.Run(() => _handlers?.Invoke(this, new DriverConnectionLossEventArgs(reason))); + } + + private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect(LossSourceStubDriver driver) + { + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, + reconnectInterval: TimeSpan.FromHours(1), + healthPollInterval: FastPoll, + commsLossTimeout: TimeSpan.FromHours(1))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + parent.ExpectMsg(TimeSpan.FromSeconds(3)).Connected.ShouldBeTrue(); + return (actor, parent); + } + + [Fact] + public async Task A_connection_loss_from_a_connected_driver_reconnects_at_once_not_after_the_watchdog() + { + var driver = new LossSourceStubDriver(); + var (_, parent) = Connect(driver); + var started = DateTime.UtcNow; + + await driver.RaiseAsync("keep-alive reported BadServerHalted"); + + parent.ExpectMsg(TimeSpan.FromSeconds(3)).Connected.ShouldBeFalse(); + (DateTime.UtcNow - started).ShouldBeLessThan(TimeSpan.FromSeconds(3), + "the watchdog is configured at an hour, so only the connection-loss signal can have done this"); + } + + /// + /// The falsification control: the same driver and actor with the event suppressed stays Connected — so the + /// Reconnecting above is the signal's doing, not something the setup would have done anyway. + /// + [Fact] + public void Without_a_raise_the_same_driver_stays_connected() + { + var driver = new LossSourceStubDriver(); + var (_, parent) = Connect(driver); + + var ticksAtConnect = driver.HealthReadCount; + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(600)); + (driver.HealthReadCount - ticksAtConnect).ShouldBeGreaterThanOrEqualTo(5, + "health-poll ticks did not run — the absence assertion above would be vacuous"); + driver.HandlerCount.ShouldBe(1, "attached, so the absence above is not merely an unwired event"); + driver.InitializeCount.ShouldBe(1, "and nothing re-initialised the driver"); + } + + /// + /// A raise while NOT Connected is ignored, not deferred: here it lands while the initial connect is still in + /// flight. Acting on it later would tear down the connection that very attempt is building. + /// + [Fact] + public async Task A_raise_while_connecting_is_ignored() + { + var driver = new LossSourceStubDriver(); + var release = new TaskCompletionSource(); + var initStarted = new TaskCompletionSource(); + driver.InitBehavior = async _ => + { + initStarted.TrySetResult(); + await release.Task; + }; + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, + reconnectInterval: TimeSpan.FromHours(1), + healthPollInterval: FastPoll, + commsLossTimeout: TimeSpan.FromHours(1))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + await initStarted.Task.WaitAsync(TimeSpan.FromSeconds(3)); + + // The raise is enqueued before the connect can complete (RaiseAsync returns after the Tell), so the mailbox + // delivers it while the actor is still Connecting. + await driver.RaiseAsync("a loss reported mid-connect"); + release.SetResult(); + + parent.ExpectMsg(TimeSpan.FromSeconds(3)).Connected.ShouldBeTrue(); + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(600)); + 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() + { + var driver = new LossSourceStubDriver(); + var (actor, parent) = Connect(driver); + driver.HandlerCount.ShouldBe(1); + + parent.Watch(actor); + Sys.Stop(actor); + // PostStop announces comms loss (03/S15) before the Terminated arrives. + parent.FishForMessage(m => m is Terminated t && t.ActorRef.Equals(actor), TimeSpan.FromSeconds(3)); + + driver.HandlerCount.ShouldBe(0, "the IDriver outlives the actor on a respawn — a leaked handler holds a dead Self"); + } + + /// A driver that does not implement the interface is simply never subscribed — the signal is opt-in. + [Fact] + public void A_driver_without_the_capability_connects_as_before() + { + var driver = new StubDriver(); + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, reconnectInterval: TimeSpan.FromHours(1), healthPollInterval: FastPoll, + commsLossTimeout: TimeSpan.FromHours(1))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + parent.ExpectMsg(TimeSpan.FromSeconds(3)).Connected.ShouldBeTrue(); + parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } +}