diff --git a/CLAUDE.md b/CLAUDE.md index fee02b58..e52ff5e9 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 dropped + subscription — 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..31d79580 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -441,6 +441,31 @@ 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 dropped (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. +- Pinned by `DriverInstanceActorConnectionLossSignalTests` (Reconnecting within seconds while the watchdog is set to an + hour; a falsification control with no raise that stays Connected; a raise mid-connect ignored; detach on stop) and + `OpcUaClientConnectionLossSignalTests` (definitive raises once, ambiguous never, re-armed after re-init). + **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..38c0de5e 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -50,7 +50,18 @@ 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 different scheme is left alone; an unparseable advertised URL falls +back to the configured one. `OpcUaClientDriver.RewriteEndpointUrl`. ### Session lifetime @@ -83,12 +94,38 @@ 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. When the window closes the driver logs ONE Information line — `post-recovery replay window closed — + dropped N late notifications across M refs after recovery` — because the per-drop line is Debug and the + `stale_notification_dropped` counter is OTel-only, which made the guard unobservable on the docker-dev rig (#676). +- **Trailing re-read after an in-place recovery (#676).** Found by the TCP-stall wire test: a change the server + published just after the stall ended rode a Publish response that had been in flight across the stall; the SDK + timed that request out (`BadTimeout`), discarded the response and never republished the message (the following + keep-alives moved the sequence number past it), so the node stayed **Good on the pre-change value indefinitely**. + `Timeout + KeepAliveInterval` after the recovery — when every request in flight across the stall has completed or + timed out — every subscribed ref is read once more, and any Good value **newer** than the last one accepted for that + ref is published (through the replay guard's mark while the window is open, so it can never regress a node). An + Information line names how many refs it had to repair. - **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 tracked subscription declared lost (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. - **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 +133,8 @@ 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 dead subscription never comes + back on its session, so this also raises `ConnectionLost` (#676) and the re-initialise happens at once. - 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 +164,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 +291,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..0eb7615b 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,12 @@ 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 dropped). Re-armed when the session + /// is released (re-initialise / shutdown). + 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 +171,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 +292,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; @@ -883,9 +912,34 @@ 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 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; + var builder = new UriBuilder(advertised) { Host = configured.DnsSafeHost, Port = configured.Port }; + return builder.Uri.ToString(); + } + /// /// Build a carrying a client user-authentication /// certificate loaded from . @@ -986,6 +1040,8 @@ 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); } /// Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream. @@ -1021,8 +1077,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 @@ -1766,6 +1827,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit 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, @@ -1798,14 +1866,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 +1882,36 @@ 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(); + } + _logger.LogInformation( + "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)) @@ -1839,6 +1925,25 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit foreach (var kv in _lastGoodSourceTicks) _replayGuard[kv.Key] = kv.Value; Interlocked.Exchange(ref _replayGuardUntilTicks, (now + window).Ticks); } + 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) + { + try + { + var wait = new DateTime(until, DateTimeKind.Utc) - Clock.GetUtcNow().UtcDateTime; + if (wait > TimeSpan.Zero) await Task.Delay(wait + TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); + if (Clock.GetUtcNow().UtcDateTime.Ticks >= until) DisarmReplayGuard(until); + } + catch (Exception ex) + { + _logger.LogDebug(ex, "OpcUaClientDriver '{DriverInstanceId}': replay-guard disarm timer faulted.", _driverInstanceId); + } } /// @@ -2683,7 +2788,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,12 +2813,15 @@ 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; + _reconcileTask = ReconcileAfterRecoveryAsync(sender, recovery, PostRecoveryReconcileDelay()); + _ = DisarmReplayGuardWhenDueAsync(guardUntil); return; } @@ -2729,14 +2839,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); + } } /// @@ -2777,8 +2929,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit 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 dead subscription never comes back on this session — only the re-subscribe after a re-initialise + // recreates it — so this is definitive: ask the host now rather than after the watchdog's 90 s. + RaiseConnectionLost($"subscription lost: {reason}"); } /// @@ -2880,6 +3035,74 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } } + /// + /// #676 — the trailing reconcile read after an in-place recovery. Found by the TCP-stall wire test: a value change + /// the server published just after a stall ended rode a Publish response that had been in flight across the + /// stall; the SDK timed that request out (BadTimeout), discarded the response, and never republished the + /// message (the following keep-alives advanced the sequence number past it). The node stayed Good on the value + /// from the recovery read — stale, indefinitely, for a value that does not change again. Once every request that + /// was in flight across the stall has completed or timed out (), every + /// subscribed ref is read once more and any Good value NEWER than the last one accepted for that ref is published + /// (through the replay guard's mark while its window is open, so it can never regress a node). + /// + /// The session that recovered; nothing happens if it is no longer current or is in outage. + /// The recovery read, which runs first so this read is compared against its result. + /// How long to wait before reading. + private async Task ReconcileAfterRecoveryAsync(ISession session, Task recovery, TimeSpan delay) + { + bool Current() => ReferenceEquals(session, Session) + && Volatile.Read(ref _sessionLost) == 0 + && Volatile.Read(ref _subscriptionLost) == 0; + try + { + if (delay > TimeSpan.Zero) await Task.Delay(delay).ConfigureAwait(false); + await recovery.ConfigureAwait(false); + if (!Current()) return; + var refs = _subscriptions.Values + .SelectMany(rs => rs.ItemHandlers.Select(p => RefOf(p.Item))) + .Distinct(StringComparer.Ordinal) + .ToList(); + if (refs.Count == 0) return; + var (results, transportOk) = await ReadCoreAsync(refs, CancellationToken.None, session).ConfigureAwait(false); + if (!transportOk) return; // a new outage — the keep-alive reports it + var published = 0; + for (var i = 0; i < refs.Count; i++) + { + if (!Current()) return; + var r = results[i]; + if (!StatusCode.IsGood(r.StatusCode) || r.SourceTimestampUtc is not { } src) continue; + if (!IsNewerThanAccepted(refs[i], src)) continue; + RecordUpstreamGood(refs[i], r); + _lastValues[refs[i]] = r; + _awaitingRecovery.TryRemove(refs[i], out _); + foreach (var rs in _subscriptions.Values) + if (rs.ItemHandlers.Any(p => RefOf(p.Item) == refs[i])) + RaiseDataChange(new DataChangeEventArgs(rs.Handle, refs[i], r)); + 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 (Exception ex) + { + _logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': trailing re-read after a recovery failed.", + _driverInstanceId); + } + } + + /// #676 — whether a Good upstream value with source time is newer than the last one + /// accepted for . Inside the replay window this IS the guard's check (and advances its mark); + /// outside it, a strict comparison against the last accepted Good source time. + private bool IsNewerThanAccepted(string fullRef, DateTime source) + { + var until = Interlocked.Read(ref _replayGuardUntilTicks); + if (until != 0 && Clock.GetUtcNow().UtcDateTime.Ticks < until) return !IsLateReplay(fullRef, source, out _); + return !_lastGoodSourceTicks.TryGetValue(fullRef, out var last) || source.Ticks > last; + } + /// Quality-only annotation (#626 pattern) of every subscribed alarm condition. private void AnnotateAlarmQuality(uint quality, DateTime now) { 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..6940964a 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,12 @@ 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, so it cannot land in Reconnecting as an unhandled + // message. + 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 +770,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(_ => { /* keep retrying via timer */ }); Receive(StoreDesiredSubscriptions); Receive(_ => { /* already reconnecting — 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 => @@ -1105,6 +1112,50 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// one), marshaling each transition to the actor thread. Idempotent; mirrors /// , including the PreStart-not-per-connect placement — a probe /// loop can report a host down while the driver itself is between connects. + /// + /// Subscribe the driver's (#676), if it is one, + /// marshalling each raise to the actor thread — actor state is never touched on the driver's thread. Attached + /// 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(); + } + private void AttachHostStatusSource() { if (_driver is not IHostConnectivityProbe src || _hostStatusHandler is not null) return; @@ -1497,6 +1548,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..109074c0 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientConnectionLossSignalTests.cs @@ -0,0 +1,408 @@ +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) + { + lock (Entries) + return [.. Entries.Where(e => e.Level == LogLevel.Information && 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); + } + + // ── 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_still_reports_it_closed_and_counts_restart_per_window() + { + 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(31)); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(2, t0.AddSeconds(40))); + + drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); + drv.RaiseKeepAliveForTest(StatusCodes.Good); + await drv.RecoveryTaskForTest; + clock.Advance(TimeSpan.FromSeconds(31)); + drv.DeliverNotificationForTest(handle, SpeedRef, Good(3, t0.AddSeconds(80))); + + log.Info("replay window closed").ShouldBe([ + "OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 1 late notifications across 1 refs after recovery.", + "OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 0 late notifications across 0 refs after recovery.", + ]); + } + + // ── 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"); + } +} 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..55a3762a --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientEndpointRewriteTests.cs @@ -0,0 +1,53 @@ +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_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..5cee9ec5 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverStallWireTests.cs @@ -0,0 +1,144 @@ +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. + await using var drv = OpcUaClientDriverWireTests.NewDriver( + rig, "wire-stall", endpointUrl: proxy.Through(rig.EndpointUrl), timeout: TimeSpan.FromSeconds(3)); + 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; + var badKeepAlives = 0; + session.KeepAlive += (_, e) => + { + if (ServiceResult.IsBad(e.Status)) Interlocked.Increment(ref 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 a further failed keep-alive (counted, not slept for). + var badAtDetection = Volatile.Read(ref badKeepAlives); + await WaitAsync(() => Volatile.Read(ref badKeepAlives) >= badAtDetection + 2, "further keep-alives must fail while stalled"); + 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)"); + 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"); + + 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..7b5709f1 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,10 @@ 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) => new(new OpcUaClientDriverOptions { - EndpointUrl = rig.EndpointUrl, + EndpointUrl = endpointUrl ?? rig.EndpointUrl, RawTags = [ Tag(SpeedRef, ServerEventRig.RawNsu(SpeedId)), @@ -58,11 +58,10 @@ public sealed class OpcUaClientDriverWireTests Username = "apikey", Password = ServerEventRig.KeyPassword, AutoAcceptCertificates = true, - Timeout = TimeSpan.FromSeconds(10), + Timeout = timeout ?? TimeSpan.FromSeconds(10), SessionTimeout = TimeSpan.FromSeconds(30), 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..e7da8d01 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConnectionLossSignalTests.cs @@ -0,0 +1,157 @@ +using Akka.Actor; +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); + } + + [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)); + } +}