Merge feat/676-opcuaclient-followups: endpoint host+port rewrite, TCP-stall wire test, ReconnectPeriod removed, IDriverConnectionLossSource fast recovery, trailing re-read, replay summary (#676)

This commit is contained in:
Joseph Doherty
2026-10-11 11:45:26 -04:00
22 changed files with 1876 additions and 117 deletions
+6
View File
@@ -321,6 +321,12 @@ non-`Healthy` state **and** its `LastSuccessfulRead` has been frozen for 90 s. T
`DefaultReconnectInterval`. Unlike the #523 drift detector this generates **no** traffic on a healthy
system — it is a local `GetHealth()` read, and only reaches the network once a driver has already stopped
reading.
- **Fast path for DEFINITIVE losses (#676):** a driver implementing the opt-in `IDriverConnectionLossSource` raises
`ConnectionLost` (once per connection) when its protocol proves the connection can never recover in place — OpcUaClient:
keep-alive `BadSessionIdInvalid`/`BadSessionClosed`/`BadSessionNotActivated`/`BadServerHalted` or a subscription the
SERVER reported dropped (its publishing-stall heuristic does not raise) — and `DriverInstanceActor` (attach PreStart, detach PostStop) turns it into the same disconnect only
while Connected; ambiguous failures (timeouts, `BadNoCommunication`, a stalled socket) must never raise and stay with
the watchdog's conjunction. `docs/DriverLifecycle.md`.
- **What comms loss now does downstream — one contract, six seams (archreview R3-02).** "A value consumer
never reports a quality better than its source's" is stated once in `docs/DriverLifecycle.md` §"The
+30
View File
@@ -441,6 +441,36 @@ Two consequences worth knowing:
**notices**, so a silent-success termination path — a stream that ends cleanly, a poll loop that exits
without throwing — bypasses it entirely.
**A driver that KNOWS its connection is gone can skip the wait: `IDriverConnectionLossSource` (Gitea #676).** The
watchdog has to prove a loss from silence, so it costs ~90 s even when the protocol already said "this connection
will never come back". An opt-in `Core.Abstractions` interface lets a driver say so:
`event EventHandler<DriverConnectionLossEventArgs> ConnectionLost` (a reason string). `DriverInstanceActor` attaches
in `PreStart` and detaches in `PostStop` — the same lifetime as `IRediscoverable`, for the same reason (the `IDriver`
outlives the actor on a respawn) — marshals each raise to the actor thread (`Self.Tell`; actor state is never touched
on the driver's thread), and acts on it **only while `Connected`**, through the very handler the watchdog's
`DisconnectObserved` uses (detach, record a fault, `Reconnecting`, teardown-then-init on the retry). In
`Connecting`/`Reconnecting`/`Stubbed` it is ignored, not deferred: a re-initialise is already pending, and acting on a
stale raise later could tear down the connection that attempt is building. The two are complementary, not
alternatives:
- **The signal is for DEFINITIVE losses only.** A driver must raise only when no in-place recovery is possible, at most
once per connection, re-armed by a re-initialise. An ambiguous failure — a timeout, a stalled socket, "no
communication" — must not raise: a stalled link that resumes keeps its session, subscriptions and in-place recovery,
and tearing it down would cost all three. Those stay with the watchdog, whose conjunction is the honest test for them.
- **Today only `OpcUaClientDriver` implements it:** keep-alive `BadSessionIdInvalid` / `BadSessionClosed` /
`BadSessionNotActivated` / `BadServerHalted`, or a subscription the server REPORTED dropped (not the driver's
publishing-stall heuristic, which is an inference from silence and stays with the watchdog; see
[OpcUaClient.md](drivers/OpcUaClient.md) §"Upstream outage and recovery" for why `BadServerHalted` is on the list —
it was measured, the SDK never re-opens a halted channel). A gracefully restarted upstream therefore recovers in
about one reconnect interval instead of ~100 s; a killed one still waits for the watchdog.
- ⚠️ **Known gap:** "only while `Connected`" plus "once per connection" means a raise the host receives while not yet
Connected (e.g. the protocol proves the fresh connection dead before `InitializeSucceeded` is processed) is swallowed,
and that connection falls back to the watchdog — the pre-#676 recovery time, in a narrow window.
- Pinned by `DriverInstanceActorConnectionLossSignalTests` (Reconnecting within seconds while the watchdog is set to an
hour; a falsification control with no raise that stays Connected; a raise mid-connect ignored; a stray watchdog
`DisconnectObserved` in `Reconnecting` is a no-op; detach on stop) and `OpcUaClientConnectionLossSignalTests`
(definitive raises once, ambiguous and the stall heuristic never, re-armed after re-init).
**The reconnect retry now tears down before re-initialising.** `Reconnecting`'s `RetryConnect` calls
`ShutdownAsync` before `InitializeAsync` — the sequence every driver's own `ReinitializeAsync` already
uses — because several drivers **overwrite** an `IAsyncDisposable` transport field inside `InitializeAsync`
+76 -8
View File
@@ -50,7 +50,19 @@ provides ordered **failover** — the driver tries each candidate in turn at eve
4840 / 4841). See
`src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs`
for every field (security policy/mode, auth type, session timeout, keep-alive,
browse root, node/depth caps). `ReconnectPeriod` is still parsed but no longer used (#670).
browse root, node/depth caps). `ReconnectPeriod` was **deleted** (#676 — unused since #670 removed the SDK reconnect
handler, and the AdminUI no longer shows it); stored configs that still carry `reconnectPeriod` keep parsing in the
factory, the Test-Connect probe and the browse picker (unknown keys are skipped), and the form drops it on the next save.
### Endpoint address: the configured host and port win (#676)
GetEndpoints returns each endpoint with the URL the **server** believes it lives at. Behind NAT, a port map, a
container port publish or a proxy that address is unreachable from the client — or reachable but bypassing the proxy.
The driver therefore rewrites the selected endpoint's **host and port** to the configured URL's whenever the schemes
match, keeping the advertised path (the SDK's own `CoreClientUtils.SelectEndpoint` rule, which the driver's
policy-aware endpoint selection used to bypass). A configured URL with **no port** keeps the advertised port (copying
the port-less `-1` would have sent the session to 4840). A different scheme is left alone; an unparseable advertised
URL falls back to the configured one. `OpcUaClientDriver.RewriteEndpointUrl`.
### Session lifetime
@@ -83,12 +95,57 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal
without moving the mark. Outside that window every notification is
accepted whatever its timestamp, so an upstream clock that steps backwards (NTP, a PLC clock reset) never
freezes a node. The outage Bad never moves the comparison; a re-initialise resets it.
⚠️ **By design, inside the window a notification whose source time EQUALS the last accepted one is dropped** — so a
status-only change that keeps the old source timestamp (not what OPC UA servers are supposed to do) is lost during
the window only (the trailing re-read below repairs it). When the window closes — or a second recovery re-arms it —
the driver logs ONE line, `post-recovery replay window closed — dropped N late notifications across M refs after
recovery`, at **Information when N > 0** and Debug otherwise, because the per-drop line is Debug and the
`stale_notification_dropped` counter is OTel-only, which made the guard unobservable on the docker-dev rig (#676).
A re-arm starts the count afresh.
- **Trailing re-read after an in-place recovery (#676).** Found by the TCP-stall wire test: a change the server
published just after the stall ended rode a Publish response that had been in flight across the stall; the SDK
timed that request out (`BadTimeout`), discarded the response and never republished the message (the following
keep-alives moved the sequence number past it), so the node stayed **Good on the pre-change value indefinitely**.
`Timeout + KeepAliveInterval` after the recovery — when every request in flight across the stall has completed or
timed out — every subscribed ref is read once more and published when its **(value, status) differs** from the last
value delivered for that ref **and is not older** than it. Any status counts: a Good→Bad/Uncertain change lost in the
pipeline is as stale as a lost value. "Older" compares source timestamps only when both carry one; a read with no
source timestamp (a server that does not send them) is the server's current state and counts as newer. An **equal**
source timestamp with different content IS published — it is the server's current state, not a replay, and it repairs
the equal-timestamp status-only change the replay guard drops by design. An identical (value, status) is never
republished. A ref that received ANY delivery while the read was in flight is skipped (a per-ref delivery counter is
captured before the read is issued): without source timestamps the read's sample cannot be ordered against that
notification, and publishing it would regress the node. An Information line names how many refs it had to repair. The trailing read and the replay-guard disarm
timer run on the driver clock and are cancelled when the session is released (re-initialise / shutdown / dispose).
- **No older value is ever delivered after a newer one.** The paths that publish upstream data — live notifications,
the recovery read, the trailing read and the comms-loss Bad fan — make their "may I publish this?" decision and the
publish inside one driver-wide critical section. A single lock rather than one per ref: it also serialises delivery
across subscriptions (the SDK dispatches each subscription's messages on its own worker), which is cheap because each
section is a dictionary update plus a subscriber `Tell` and nothing awaits under it. Subscribers run inside it — the
host's handler only `Tell`s an actor — so a subscriber must never block on, or call back into, the driver.
- **The session does not come back** (an upstream restart: the server forgot the session) ⇒ the driver stays
`Degraded` with its read stamp frozen, which is exactly the #527 comms-loss watchdog's trip condition. After 90 s
`DriverInstanceActor` runs `ShutdownAsync` + `InitializeAsync` (retried every 10 s while the server is down) and
re-subscribes data + alarms; the new subscription's initial values come back Good. Alarm last-told state survives
the re-initialise (#619), so the post-subscribe `ConditionRefresh` raises no un-ack flicker; the host's reconnect
annotation restores condition quality.
`Degraded` with its read stamp frozen, which is the #527 comms-loss watchdog's trip condition; `DriverInstanceActor`
runs `ShutdownAsync` + `InitializeAsync` (retried every 10 s while the server is down) and re-subscribes data +
alarms; the new subscription's initial values come back Good. Alarm last-told state survives the re-initialise
(#619), so the post-subscribe `ConditionRefresh` raises no un-ack flicker; the host's reconnect annotation restores
condition quality.
- **Fast recovery for a DEFINITIVELY lost session (#676).** The driver implements `IDriverConnectionLossSource` and
raises `ConnectionLost` — **once per session**, re-armed by a re-initialise — when the loss is certain, so the host
re-initialises at once instead of after the watchdog's 90 s. Definitive means: a keep-alive status of
`BadSessionIdInvalid` / `BadSessionClosed` / `BadSessionNotActivated` (the server forgot, closed or never activated
the session) or `BadServerHalted`, or a subscription the **server** reported gone (below). `BadServerHalted` is on the list
because it was **measured**: on the restart wire test a gracefully stopped server reports `BadServerHalted` (and the
keep-alives alternate it with `BadNoCommunication`), and the SDK never re-opens a channel the server halted — 15 s
after the server was back, the same session still failed every request with `BadServerHalted`; `BadSessionIdInvalid`
never arrived. Everything else — `BadNoCommunication`, `BadConnectionClosed`, `BadNotConnected`,
`BadRequestTimeout`/`BadTimeout` (what a stalled link produces) — is **ambiguous** (the link may resume and keep its
session, see the stall test) and stays with the watchdog. Consequence: a gracefully restarted upstream recovers in
roughly one reconnect interval (≈10 s) instead of ~100 s; an upstream that vanishes without a halt (killed, cable
pulled) still waits for the watchdog. ⚠️ **Known gap:** the host acts on the signal only while the driver is
`Connected`, and the driver raises it once per session — so a definitive loss raised while the host is not
Connected (e.g. `BadServerHalted` on the fresh session before the host has processed the successful initialise) is
swallowed and that session falls back to the watchdog. Re-raising on later keep-alives would close it but was left out
to keep the once-per-session contract simple; the cost is the pre-#676 recovery time, in a narrow window.
- **The subscription dies but the session lives** (an outage longer than the subscription lifetime —
`LifetimeCount` x publishing interval, often well under the session timeout — or a server that drops it) ⇒ keep-alives
stay Good, so they cannot be the signal. A `PublishError` of `BadNoSubscription` (while subscriptions are tracked) or
@@ -96,7 +153,10 @@ The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal
persists for one more subscription keep-alive window + 5 s across good keep-alives, declares it lost: `Degraded`
(`"subscription lost: …"`), the outage published once, and — sticky until the session is released — **nothing**
advances `LastSuccessfulRead` any more (not good keep-alives, not a sibling subscription's notifications), so the
watchdog's re-initialise recreates it. Both detectors are wire-tested independently.
watchdog's re-initialise recreates it. Both detectors are wire-tested independently. A subscription the **server**
reported gone (`BadNoSubscription` / `BadSubscriptionIdInvalid`) never comes back on its session, so that also raises
`ConnectionLost` (#676) and the re-initialise happens at once; the `PublishingStopped` **heuristic** is an inference
from silence and keeps going through the watchdog.
- An in-place recovery is bound to the session whose keep-alive recovered: one overtaken by a re-initialise neither
publishes, nor degrades, nor re-arms the outage latch of the new session.
@@ -126,7 +186,7 @@ initialise (the AdminUI form drops them on the next save). Pinned by
## Capability surface
`OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource`
`OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, IDriverConnectionLossSource`
(`src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs:31`).
| Capability | Path | Notes |
@@ -253,6 +313,14 @@ the reconnect handler, #670). They are not part of `DriverHealth` or the
- **Integration fixture** — exercises the driver against a reference OPC UA
server (opc-plc) on the shared docker host; see
[OpcUaClient-Test-Fixture.md](OpcUaClient-Test-Fixture.md) for the coverage map.
- **Wire tests against the in-process server** — `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/`
`OpcUaClientDriver*WireTests` (no Docker). `OpcUaClientDriverStallWireTests` (#676) covers the in-place blip that
used to be live-gate-only: the driver connects through `TcpStallProxy`, an in-test TCP forwarder whose pause holds
every byte without closing anything (which only works because of the endpoint rewrite above); it asserts one Bad,
a frozen read stamp, recovery on the SAME session, the identical value republished Good, a change published right
after recovery still arriving, and no source-time regression once the SDK's late backlog flush has arrived.
`OpcUaClientDriverReconnectWireTests` covers the server restart (incl. the `BadServerHalted` → `ConnectionLost` signal)
and the subscription-lost case.
## Further reading
-1
View File
@@ -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 |
@@ -0,0 +1,31 @@
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// <summary>
/// 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
/// <see cref="ConnectionLost"/> so the host re-initialises it at once instead of waiting out the #527 comms-loss
/// watchdog (~90 s of frozen <see cref="DriverHealth.LastSuccessfulRead"/>).
/// </summary>
/// <remarks>
/// <para>
/// <b>Definitive only.</b> An AMBIGUOUS failure — a timeout, a stalled socket, "no communication" — must NOT
/// raise: the connection may yet resume in place (a <c>docker pause</c>, 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.
/// </para>
/// <para>
/// The host (<c>DriverInstanceActor</c>) 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 <see cref="IDriver"/> instance can outlive the actor. A driver should raise at most
/// once per connection and re-arm when it is re-initialised.
/// </para>
/// </remarks>
public interface IDriverConnectionLossSource
{
/// <summary>Raised (on the driver's own thread) when the driver's connection is definitively lost.</summary>
event EventHandler<DriverConnectionLossEventArgs>? ConnectionLost;
}
/// <summary>Event payload for <see cref="IDriverConnectionLossSource.ConnectionLost"/>.</summary>
/// <param name="Reason">Driver-supplied reason for the diagnostic log (e.g. the protocol status that proved the loss).</param>
public sealed record DriverConnectionLossEventArgs(string Reason);
@@ -101,12 +101,6 @@ public sealed record OpcUaClientDriverOptions
/// <summary>Client-side keep-alive interval.</summary>
public TimeSpan KeepAliveInterval { get; init; } = TimeSpan.FromSeconds(5);
/// <summary>
/// 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.
/// </summary>
public TimeSpan ReconnectPeriod { get; init; } = TimeSpan.FromSeconds(5);
/// <summary>
/// When <c>true</c>, the driver accepts any self-signed / untrusted server certificate.
/// Dev-only — must be <c>false</c> in production so MITM attacks against the opc.tcp
@@ -29,7 +29,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
/// back on its own recovers in place.
/// </para>
/// </remarks>
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
{
/// <summary>Authored tags whose TagConfig carried no usable <c>nodeId</c>, so there is no upstream node
/// to dial (Gitea #541). This driver's skip path was the quietest of the twelve: <c>NamespaceMap</c>
@@ -83,6 +83,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_logger = logger ?? NullLogger<OpcUaClientDriver>.Instance;
_secretResolver = secretResolver ?? NullSecretResolver.Instance;
_sdkTelemetry = logger is null ? SdkTelemetry.Context : SdkTelemetry.ForLogger(logger);
PostRecoveryReconcileDelay = () => _options.Timeout + _options.KeepAliveInterval;
}
/// <summary>Mutable so <see cref="InitializeAsync"/> can adopt a re-parsed config on reinitialize
@@ -135,6 +136,14 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>Occurs when the host connectivity status of the remote OPC UA server changes.</summary>
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
/// <inheritdoc />
/// <remarks>Gitea #676 — raised at most once per session, and only on a DEFINITIVE loss
/// (<see cref="IsDefinitiveSessionLoss"/>, or a tracked subscription the SERVER reported dropped — never the
/// publishing-stall heuristic). Re-armed when the session is released (re-initialise / shutdown). Known gap: the host
/// acts on it only while Connected, so a raise it receives before it has processed the successful initialise is
/// swallowed and that session falls back to the #527 watchdog.</remarks>
public event EventHandler<DriverConnectionLossEventArgs>? ConnectionLost;
// OPC UA StatusCode constants the driver surfaces for local-side faults. Upstream-server
// StatusCodes are passed through verbatim per driver-specs.md §8 "cascading quality" —
// downstream clients need to distinguish 'remote source down' from 'local driver failure'.
@@ -164,6 +173,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>Test-only seam (#670): the most recent in-place recovery task.</summary>
internal Task RecoveryTaskForTest => _recoveryTask;
/// <summary>Test seam (#676): the trailing reconcile read scheduled by the last in-place recovery.</summary>
internal Task ReconcileTaskForTest => _reconcileTask;
/// <summary>Test-only seam (#670 review M1): raise the session's PublishError as the SDK publish loop does.</summary>
/// <param name="status">The publish error status.</param>
/// <param name="subscriptionId">The subscription id the error names (0 = none).</param>
@@ -282,6 +294,25 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>The last in-place recovery started (#670). Test seam: lets a test await it deterministically.</summary>
private Task _recoveryTask = Task.CompletedTask;
/// <summary>#676 — 1 once <see cref="ConnectionLost"/> 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.</summary>
private int _connectionLostRaised;
/// <summary>#676 — the trailing reconcile read scheduled by the last in-place recovery. Test seam.</summary>
private Task _reconcileTask = Task.CompletedTask;
/// <summary>#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 <see cref="_replayGuardLock"/>.</summary>
private int _replayDrops;
private readonly HashSet<string> _replayDropRefs = new(StringComparer.Ordinal);
/// <summary>
/// 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.
/// </summary>
internal Func<TimeSpan> PostRecoveryReconcileDelay { get; set; }
/// <summary>Test seam (#670 review M1): whether a subscription has stopped publishing. Production asks the SDK.</summary>
internal Func<Subscription, bool> IsPublishingStopped { get; set; } = static s => s.Created && s.PublishingStopped;
@@ -326,6 +357,37 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>Serialises arming against expiry, so an expiry racing a re-arm cannot wipe the new seeds.</summary>
private readonly object _replayGuardLock = new();
/// <summary>
/// Review of 3b88056f — serialises "decide whether this upstream value may be published" with "publish it" across
/// the three paths that publish upstream data: live notifications, the recovery read and the trailing reconcile
/// read. Without it a path could pass its newer-than check, lose the CPU to another path publishing a NEWER value,
/// and then deliver its older one last — regressing the node. It also covers the comms-loss Bad fan
/// (<see cref="PublishCommsLoss"/>). One driver-wide lock rather than one per ref: it also serialises delivery
/// ACROSS subscriptions — with several subscriptions the SDK dispatches each subscription's messages on its own
/// worker, so their notifications no longer run in parallel. That is the cost, and it is small: each critical
/// section is a dictionary update plus a subscriber call that only <c>Tell</c>s an actor (tens of nanoseconds to
/// a few microseconds), nothing awaits under it, and per-subscription message order is unaffected. Because
/// subscribers are invoked under it, a subscriber must never block on, or call back into, this driver.
/// </summary>
private readonly Lock _deliveryLock = new();
/// <summary>
/// Re-review of 30a9d5f0 — per RawPath, a counter bumped (under <see cref="_deliveryLock"/>) every time a value is
/// delivered for it. The trailing re-read captures it before issuing its read and skips any ref delivered to since:
/// without source timestamps the read's value cannot be ordered against a notification that landed while the read
/// was in flight, and publishing it would regress the node to the value the read sampled.
/// </summary>
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, long> _deliverySeq =
new(StringComparer.Ordinal);
/// <summary>Review of 3b88056f — cancels the post-recovery background work (trailing read, replay-guard disarm timer)
/// of the current session; cancelled and replaced whenever the session is released, and on dispose.</summary>
private CancellationTokenSource _sessionWorkCts = new();
/// <summary>Test seam (review of 3b88056f): invoked inside the trailing read's critical section, after it decided
/// to publish a ref and before it does — where a racing notification would otherwise overtake it.</summary>
internal Action<string>? BeforeReconcilePublishForTest { get; set; }
/// <summary>Floor of the replay-guard window; the live gate saw the late replay 4–5 s after recovery.</summary>
private static readonly TimeSpan ReplayGuardMinimum = TimeSpan.FromSeconds(30);
@@ -883,9 +945,40 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
$"No endpoint at '{endpointUrl}' matches SecurityPolicy={policy} + SecurityMode={mode}. " +
$"Server advertises: {advertised}");
}
// #676: connect where the operator pointed us, not where the server says it lives.
match.EndpointUrl = RewriteEndpointUrl(match.EndpointUrl, endpointUrl);
return match;
}
/// <summary>
/// Gitea #676 — the address to open the session on. A server reports its endpoints with ITS view of its own
/// address; behind NAT, a port map, a container port publish or a proxy that address is unreachable from here
/// (or bypasses the proxy). When the advertised URL's scheme matches the configured one, the configured host
/// AND port (when it names one) replace the advertised ones and the advertised path is kept — the SDK's own
/// <c>CoreClientUtils.SelectEndpoint</c> 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.
/// </summary>
/// <param name="advertisedUrl">The <c>EndpointUrl</c> the server returned from GetEndpoints.</param>
/// <param name="configuredUrl">The URL the driver was configured with (the one GetEndpoints was sent to).</param>
/// <returns>The URL to connect the session to.</returns>
internal static string RewriteEndpointUrl(string? advertisedUrl, string configuredUrl)
{
if (!Uri.TryCreate(configuredUrl, UriKind.Absolute, out var configured)) return advertisedUrl ?? configuredUrl;
if (string.IsNullOrWhiteSpace(advertisedUrl)
|| !Uri.TryCreate(advertisedUrl, UriKind.Absolute, out var advertised))
return configuredUrl;
if (!string.Equals(advertised.Scheme, configured.Scheme, StringComparison.OrdinalIgnoreCase)) return advertisedUrl;
// A port-less configured URL (Uri.Port == -1 for opc.tcp) keeps the advertised port — copying -1 would drop it
// and the SDK would fall back to 4840 (review of 3b88056f; same fix as the client's EndpointSelector).
var builder = new UriBuilder(advertised)
{
Host = configured.DnsSafeHost,
Port = configured.Port > 0 ? configured.Port : advertised.Port,
};
return builder.Uri.ToString();
}
/// <summary>
/// Build a <see cref="UserIdentity"/> carrying a client user-authentication
/// certificate loaded from <see cref="OpcUaClientDriverOptions.UserCertificatePath"/>.
@@ -986,6 +1079,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
Interlocked.Exchange(ref _sessionLost, 0);
Interlocked.Exchange(ref _subscriptionLost, 0);
Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0);
// #676: re-arm the definitive-loss signal for the next session.
Interlocked.Exchange(ref _connectionLostRaised, 0);
// Review of 3b88056f: the released session's post-recovery timers must not outlive it.
CancelSessionWork();
}
/// <summary>Cancel the current session's background work and start a fresh token for the next one.</summary>
private void CancelSessionWork()
{
// Cancelled but deliberately NOT disposed (re-review of 30a9d5f0): the keep-alive thread may be reading the old
// source's Token concurrently, and a disposed source throws ObjectDisposedException there. Nothing here is
// timer-backed (no CancelAfter), so an undisposed source holds no unmanaged resource.
var old = Interlocked.Exchange(ref _sessionWorkCts, new CancellationTokenSource());
old.Cancel();
}
/// <summary>Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream.</summary>
@@ -1021,8 +1128,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// previous connection's value bookkeeping applies to them.
_lastValues.Clear();
_lastGoodSourceTicks.Clear();
_replayGuard.Clear();
Interlocked.Exchange(ref _replayGuardUntilTicks, 0);
lock (_replayGuardLock)
{
_replayGuard.Clear();
Interlocked.Exchange(ref _replayGuardUntilTicks, 0);
_replayDrops = 0;
_replayDropRefs.Clear();
}
_awaitingRecovery.Clear();
// #619 re-review: the last-told alarm states (and ack targets) deliberately SURVIVE a shutdown. A
// ReinitializeAsync (comms-loss watchdog / Force Reconnect) is followed by the host re-subscribing the
@@ -1762,20 +1874,33 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
StatusCode: dv.StatusCode.Code,
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? DateTime.UtcNow : dv.ServerTimestamp);
// F2: inside the post-recovery replay window only, a late OLDER notification must not regress the node.
if (IsLateReplay(fullRef, snapshot.SourceTimestampUtc, out var newest))
// Review of 3b88056f: the replay check and the publish are one critical section shared with the recovery and
// trailing reads, so an older value can never be delivered after a newer one (see _deliveryLock).
lock (_deliveryLock)
{
_diag.Record(OpcUaClientDiagnostics.StaleNotificationDropped);
_logger.LogDebug(
"OpcUaClientDriver '{DriverInstanceId}': dropped a late replayed notification for {Ref} after a recovery " +
"(source {Source:O} is not newer than {Newest:O}).", _driverInstanceId, fullRef, snapshot.SourceTimestampUtc,
new DateTime(newest, DateTimeKind.Utc));
return;
// F2: inside the post-recovery replay window only, a late OLDER notification must not regress the node.
if (IsLateReplay(fullRef, snapshot.SourceTimestampUtc, out var newest))
{
_diag.Record(OpcUaClientDiagnostics.StaleNotificationDropped);
// #676: the OTel counter and the Debug line below are invisible on a default-configured node; the
// per-recovery Information summary logged at disarm is what makes the guard observable.
lock (_replayGuardLock)
{
_replayDrops++;
_replayDropRefs.Add(fullRef);
}
_logger.LogDebug(
"OpcUaClientDriver '{DriverInstanceId}': dropped a late replayed notification for {Ref} after a recovery " +
"(source {Source:O} is not newer than {Newest:O}).", _driverInstanceId, fullRef, snapshot.SourceTimestampUtc,
new DateTime(newest, DateTimeKind.Utc));
return;
}
RecordUpstreamGood(fullRef, snapshot);
_lastValues[fullRef] = snapshot;
_deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1);
_awaitingRecovery.TryRemove(fullRef, out _);
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot));
}
RecordUpstreamGood(fullRef, snapshot);
_lastValues[fullRef] = snapshot;
_awaitingRecovery.TryRemove(fullRef, out _);
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullRef, snapshot));
}
/// <summary>#670 F2 — remember the last accepted Good upstream source time for <paramref name="fullRef"/>.</summary>
@@ -1798,14 +1923,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (until == 0) return false;
if (Clock.GetUtcNow().UtcDateTime.Ticks >= until)
{
lock (_replayGuardLock)
{
if (Interlocked.Read(ref _replayGuardUntilTicks) == until)
{
Interlocked.Exchange(ref _replayGuardUntilTicks, 0);
_replayGuard.Clear();
}
}
DisarmReplayGuard(until);
return false;
}
if (source is not { } src) return false;
@@ -1821,11 +1939,40 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
}
/// <summary>
/// Expire the replay guard armed until <paramref name="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.
/// </summary>
/// <param name="until">The expiry the caller observed; only that arming is disarmed.</param>
private void DisarmReplayGuard(long until)
{
int drops, refs;
lock (_replayGuardLock)
{
if (Interlocked.Read(ref _replayGuardUntilTicks) != until) return;
Interlocked.Exchange(ref _replayGuardUntilTicks, 0);
_replayGuard.Clear();
drops = _replayDrops;
refs = _replayDropRefs.Count;
_replayDrops = 0;
_replayDropRefs.Clear();
}
LogReplaySummary(drops, refs);
}
/// <summary>One line per replay window: Information when the guard dropped something, Debug otherwise.</summary>
private void LogReplaySummary(int drops, int refs) =>
_logger.Log(drops > 0 ? LogLevel.Information : LogLevel.Debug,
"OpcUaClientDriver '{DriverInstanceId}': post-recovery replay window closed — dropped {Drops} late " +
"notifications across {Refs} refs after recovery.", _driverInstanceId, drops, refs);
/// <summary>
/// #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).
/// </summary>
private void ArmReplayGuard(DateTime now)
/// <returns>The window's expiry in clock ticks.</returns>
private long ArmReplayGuard(DateTime now)
{
var window = ReplayGuardMinimum;
foreach (var sub in _subscriptions.Values.Select(r => r.Subscription))
@@ -1833,12 +1980,39 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
var keepAliveWindow = TimeSpan.FromMilliseconds(sub.CurrentPublishingInterval * (double)sub.CurrentKeepAliveCount);
if (keepAliveWindow * 2 > window) window = keepAliveWindow * 2;
}
int supersededDrops, supersededRefs;
lock (_replayGuardLock)
{
// A re-arm inside an open window ends that window: report what it dropped and count afresh.
supersededDrops = _replayDrops;
supersededRefs = _replayDropRefs.Count;
_replayDrops = 0;
_replayDropRefs.Clear();
_replayGuard.Clear();
foreach (var kv in _lastGoodSourceTicks) _replayGuard[kv.Key] = kv.Value;
Interlocked.Exchange(ref _replayGuardUntilTicks, (now + window).Ticks);
}
if (supersededDrops > 0) LogReplaySummary(supersededDrops, supersededRefs);
return (now + window).Ticks;
}
/// <summary>
/// #676 — disarm the replay guard (and log its summary) when its window ends, even if no notification arrives to
/// disarm it lazily in <see cref="IsLateReplay"/> — a quiet subscription would otherwise never log the summary.
/// </summary>
private async Task DisarmReplayGuardWhenDueAsync(long until, CancellationToken ct)
{
try
{
var wait = new DateTime(until, DateTimeKind.Utc) - Clock.GetUtcNow().UtcDateTime;
if (wait > TimeSpan.Zero) await Task.Delay(wait + TimeSpan.FromMilliseconds(100), Clock, ct).ConfigureAwait(false);
if (Clock.GetUtcNow().UtcDateTime.Ticks >= until) DisarmReplayGuard(until);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { /* session released */ }
catch (Exception ex)
{
_logger.LogDebug(ex, "OpcUaClientDriver '{DriverInstanceId}': replay-guard disarm timer faulted.", _driverInstanceId);
}
}
/// <summary>
@@ -2683,7 +2857,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// session-recreate path cloned subscriptions the driver never re-tracked, and it kept health Healthy while
/// it did so — the #670 symptom. A session that does not come back keeps health Degraded with the read stamp
/// frozen, which is exactly what the host's #527 comms-loss watchdog trips on: ShutdownAsync + InitializeAsync
/// + re-subscribe, whose initial values come back Good.
/// + re-subscribe, whose initial values come back Good. When the keep-alive status proves the session is gone
/// for good (<see cref="IsDefinitiveSessionLoss"/>) the driver also raises <see cref="ConnectionLost"/> (#676),
/// so the host runs that sequence at once instead of after the watchdog's silence window.
/// </remarks>
private void OnKeepAlive(ISession sender, KeepAliveEventArgs e)
{
@@ -2706,18 +2882,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
UpdateHealth(_ => new DriverHealth(DriverState.Healthy, now, null));
// A fresh grace: every subscription reports stopped right after an outage until its next message.
Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0);
ArmReplayGuard(now);
var guardUntil = ArmReplayGuard(now);
_diag.Record(OpcUaClientDiagnostics.KeepAliveRecovered);
_logger.LogInformation(
"OpcUaClientDriver '{DriverInstanceId}': upstream keep-alive recovered on the same session; " +
"re-reading the refs published Bad during the outage.", _driverInstanceId);
_recoveryTask = Task.Run(() => RecoverAfterOutageAsync(sender));
var recovery = Task.Run(() => RecoverAfterOutageAsync(sender));
_recoveryTask = recovery;
var sessionWork = _sessionWorkCts.Token;
_reconcileTask = ReconcileAfterRecoveryAsync(sender, recovery, PostRecoveryReconcileDelay(), sessionWork);
_ = DisarmReplayGuardWhenDueAsync(guardUntil, sessionWork);
return;
}
if (FindStalledSubscription(now) is { } stalled)
{
DeclareSubscriptionLost($"subscription '{stalled}' stopped publishing while the session stayed up", now);
DeclareSubscriptionLost($"subscription '{stalled}' stopped publishing while the session stayed up", now,
serverReported: false);
return;
}
@@ -2729,14 +2910,56 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
TransitionTo(HostState.Stopped);
_diag.Record(OpcUaClientDiagnostics.KeepAliveMiss);
UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"keep-alive: {e.Status}"));
if (Interlocked.Exchange(ref _sessionLost, 1) == 1) return;
if (Volatile.Read(ref _subscriptionLost) == 1) return; // already published Bad for this connection
// #676: the first bad keep-alive is often ambiguous and a definitive one follows, so this is checked on EVERY
// bad keep-alive — but only after the outage below has been published, so the Bad reaches subscribers first.
var keepAliveCode = e.Status?.StatusCode ?? StatusCodes.Good;
var definitive = IsDefinitiveSessionLoss(keepAliveCode);
// A subscription loss already published Bad for this connection.
if (Interlocked.Exchange(ref _sessionLost, 1) == 0 && Volatile.Read(ref _subscriptionLost) == 0)
{
_logger.LogWarning(
"OpcUaClientDriver '{DriverInstanceId}': upstream session keep-alive failed ({Status}); subscribed values " +
"are BadNoCommunication until it recovers or the host's comms-loss watchdog reconnects the driver.",
_driverInstanceId, e.Status);
PublishCommsLoss(now);
}
if (definitive)
RaiseConnectionLost($"upstream session keep-alive reported {StatusCodes.GetBrowseName(keepAliveCode.Code)}");
}
/// <summary>
/// #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
/// (<c>BadSessionIdInvalid</c> — what a re-opened channel to a restarted server answers), closed it
/// (<c>BadSessionClosed</c>) or never activated it (<c>BadSessionNotActivated</c>); or the server HALTED
/// (<c>BadServerHalted</c>) — 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
/// <c>BadServerHalted</c>, so a halted session is dead for good. Everything else — <c>BadNoCommunication</c>,
/// <c>BadConnectionClosed</c>, <c>BadNotConnected</c>, request timeouts — is AMBIGUOUS (a stalled link that
/// resumes keeps its session, see the stall wire test) and stays with the watchdog.
/// </summary>
/// <param name="status">A keep-alive status.</param>
/// <returns>True when the session is definitively gone.</returns>
internal static bool IsDefinitiveSessionLoss(StatusCode status) =>
status == StatusCodes.BadSessionIdInvalid
|| status == StatusCodes.BadSessionClosed
|| status == StatusCodes.BadSessionNotActivated
|| status == StatusCodes.BadServerHalted;
/// <summary>#676 — raise <see cref="ConnectionLost"/> once for the current session, isolating a faulting subscriber.</summary>
/// <param name="reason">Why the connection is definitively gone, for the host's log.</param>
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);
}
}
/// <summary>
@@ -2771,14 +2994,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// frozen (sticky until the session is released), and the outage published once, so the host's watchdog
/// re-initialises and re-subscribes.
/// </summary>
private void DeclareSubscriptionLost(string reason, DateTime now)
/// <param name="reason">Why, for the log and health.</param>
/// <param name="now">Detection time.</param>
/// <param name="serverReported">True when the SERVER said the subscription is gone (BadNoSubscription /
/// BadSubscriptionIdInvalid) — definitive, so <see cref="ConnectionLost"/> is raised. False for the publishing-stall
/// heuristic, an inference from silence that stays with the #527 watchdog (review of 3b88056f).</param>
private void DeclareSubscriptionLost(string reason, DateTime now, bool serverReported)
{
if (Interlocked.Exchange(ref _subscriptionLost, 1) == 1) return;
UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"subscription lost: {reason}"));
_logger.LogWarning(
"OpcUaClientDriver '{DriverInstanceId}': {Reason}; subscribed values are BadNoCommunication until the " +
"host's comms-loss watchdog re-initialises the driver and re-subscribes.", _driverInstanceId, reason);
"host re-initialises the driver and re-subscribes.", _driverInstanceId, reason);
if (Volatile.Read(ref _sessionLost) == 0) PublishCommsLoss(now);
// #676: a subscription the SERVER says is gone never comes back on this session — only the re-subscribe after a
// re-initialise recreates it — so ask the host now rather than after the watchdog's 90 s. The stall heuristic is
// an inference, not a server statement, and keeps going through the watchdog.
if (serverReported) RaiseConnectionLost($"subscription lost: {reason}");
}
/// <summary>
@@ -2801,7 +3033,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
else
return;
if (tracked)
DeclareSubscriptionLost($"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime);
DeclareSubscriptionLost($"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime,
serverReported: true);
}
/// <summary>
@@ -2811,18 +3044,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// </summary>
private void PublishCommsLoss(DateTime now)
{
foreach (var rs in _subscriptions.Values)
// Under the delivery lock (re-review of 30a9d5f0) so the Bad cannot interleave with a recovery publish.
lock (_deliveryLock)
{
foreach (var pair in rs.ItemHandlers)
foreach (var rs in _subscriptions.Values)
{
var fullRef = RefOf(pair.Item);
_lastValues.TryGetValue(fullRef, out var last);
_awaitingRecovery[fullRef] = 0;
// F1: stamped with the DETECTION time, as the host's own comms-loss fan does. The recorder historizes
// SourceTimestampUtc ?? ServerTimestampUtc, so the last Good's source time here put the Bad at the
// Good row's (TagId, SourceTimeUtc) key, where the insert-if-absent historian dropped it.
RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, new DataValueSnapshot(
last?.Value, StatusBadNoCommunication, now, now)));
foreach (var pair in rs.ItemHandlers)
{
var fullRef = RefOf(pair.Item);
_lastValues.TryGetValue(fullRef, out var last);
_awaitingRecovery[fullRef] = 0;
_deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1);
// F1: stamped with the DETECTION time, as the host's own comms-loss fan does. The recorder historizes
// SourceTimestampUtc ?? ServerTimestampUtc, so the last Good's source time here put the Bad at the
// Good row's (TagId, SourceTimeUtc) key, where the insert-if-absent historian dropped it.
RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, new DataValueSnapshot(
last?.Value, StatusBadNoCommunication, now, now)));
}
}
}
AnnotateAlarmQuality(StatusBadNoCommunication, now);
@@ -2856,15 +3094,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
for (var i = 0; i < refs.Count; i++)
{
if (!Current()) return;
if (!_awaitingRecovery.TryRemove(refs[i], out _)) continue; // a notification got there first
_lastValues[refs[i]] = results[i];
RecordUpstreamGood(refs[i], results[i]);
// The read is upstream data published inside the replay window: a replay older than it is late too.
if (results[i].SourceTimestampUtc is { } readAt && StatusCode.IsGood(results[i].StatusCode))
_replayGuard.AddOrUpdate(refs[i], readAt.Ticks, (_, prior) => Math.Max(prior, readAt.Ticks));
foreach (var rs in _subscriptions.Values)
if (rs.ItemHandlers.Any(p => RefOf(p.Item) == refs[i]))
RaiseDataChange(new DataChangeEventArgs(rs.Handle, refs[i], results[i]));
// One critical section with the live notification path (_deliveryLock): a notification that
// removed the ref first wins, and none can overtake this publish once it has decided.
lock (_deliveryLock)
{
if (!_awaitingRecovery.TryRemove(refs[i], out _)) continue; // a notification got there first
PublishUpstreamLocked(refs[i], results[i]);
}
}
}
@@ -2880,6 +3116,108 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
}
/// <summary>
/// Publish an upstream value read (not notified) for <paramref name="fullRef"/> to every subscription carrying it,
/// recording it as the last delivered value and — when Good with a source time — as the replay guard's mark, so a
/// replay older than the read is late too. Caller holds <see cref="_deliveryLock"/>.
/// </summary>
private void PublishUpstreamLocked(string fullRef, DataValueSnapshot value)
{
_lastValues[fullRef] = value;
_deliverySeq.AddOrUpdate(fullRef, 1, (_, n) => n + 1);
RecordUpstreamGood(fullRef, value);
if (value.SourceTimestampUtc is { } readAt && StatusCode.IsGood(value.StatusCode)
&& Interlocked.Read(ref _replayGuardUntilTicks) != 0)
_replayGuard.AddOrUpdate(fullRef, readAt.Ticks, (_, prior) => Math.Max(prior, readAt.Ticks));
foreach (var rs in _subscriptions.Values)
if (rs.ItemHandlers.Any(p => RefOf(p.Item) == fullRef))
RaiseDataChange(new DataChangeEventArgs(rs.Handle, fullRef, value));
}
/// <summary>
/// #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 (<c>BadTimeout</c>), discarded the response, and never republished the
/// message (the following keep-alives advanced the sequence number past it). The node stayed on the value from
/// the recovery read — stale, indefinitely, for a value that does not change again. Once every request that was
/// in flight across the stall has completed or timed out (<see cref="PostRecoveryReconcileDelay"/>), every
/// subscribed ref is read once more and published when <see cref="IsReconcileRepair"/> says the read differs from
/// what was last delivered and is not older than it. The decision and the publish are one critical section with
/// the live notification path (<see cref="_deliveryLock"/>), so a newer notification can never be overtaken.
/// Bound to the session's work token: a re-initialise or shutdown cancels it.
/// </summary>
/// <param name="session">The session that recovered; nothing happens if it is no longer current or is in outage.</param>
/// <param name="recovery">The recovery read, which runs first so this read is compared against its result.</param>
/// <param name="delay">How long to wait before reading (on <see cref="Clock"/>).</param>
/// <param name="ct">The session's background-work token.</param>
private async Task ReconcileAfterRecoveryAsync(ISession session, Task recovery, TimeSpan delay, CancellationToken ct)
{
bool Current() => !ct.IsCancellationRequested
&& ReferenceEquals(session, Session)
&& Volatile.Read(ref _sessionLost) == 0
&& Volatile.Read(ref _subscriptionLost) == 0;
try
{
if (delay > TimeSpan.Zero) await Task.Delay(delay, Clock, ct).ConfigureAwait(false);
await recovery.WaitAsync(ct).ConfigureAwait(false);
if (!Current()) return;
var refs = _subscriptions.Values
.SelectMany(rs => rs.ItemHandlers.Select(p => RefOf(p.Item)))
.Distinct(StringComparer.Ordinal)
.ToList();
if (refs.Count == 0) return;
// Captured BEFORE the read is issued: a ref delivered to while the read is in flight has a value at least as
// new as anything the read could have sampled, so the read must not overwrite it.
var seqAtIssue = refs.Select(r => _deliverySeq.GetValueOrDefault(r)).ToArray();
var (results, transportOk) = await ReadCoreAsync(refs, ct, session).ConfigureAwait(false);
if (!transportOk) return; // a new outage — the keep-alive reports it
var published = 0;
for (var i = 0; i < refs.Count; i++)
{
if (!Current()) return;
lock (_deliveryLock)
{
if (_deliverySeq.GetValueOrDefault(refs[i]) != seqAtIssue[i]) continue;
if (!IsReconcileRepair(refs[i], results[i])) continue;
BeforeReconcilePublishForTest?.Invoke(refs[i]);
_awaitingRecovery.TryRemove(refs[i], out _);
PublishUpstreamLocked(refs[i], results[i]);
}
published++;
}
if (published > 0)
_logger.LogInformation(
"OpcUaClientDriver '{DriverInstanceId}': trailing re-read after recovery found {Count} of {Total} refs " +
"whose latest change never arrived through the subscription; published them.",
_driverInstanceId, published, refs.Count);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { /* session released */ }
catch (Exception ex)
{
_logger.LogWarning(ex, "OpcUaClientDriver '{DriverInstanceId}': trailing re-read after a recovery failed.",
_driverInstanceId);
}
}
/// <summary>
/// Whether the trailing read's <paramref name="read"/> repairs <paramref name="fullRef"/>: its (value, status)
/// differs from the last value delivered for the ref, and it is not OLDER than it. Any status counts — a Good→Bad
/// or Good→Uncertain change lost in the stalled pipeline is as stale as a lost value. "Older" is judged on source
/// timestamps only when both carry one; a read with no source timestamp (a server that does not send them) is the
/// server's CURRENT state, read after the stalled pipeline drained, so it counts as newer. An EQUAL source
/// timestamp with different content is published too: it is not a replay but the server's current state, and it
/// repairs the one change the replay guard drops by design (a status-only change that kept its source time inside
/// the window). Caller holds <see cref="_deliveryLock"/>.
/// </summary>
private bool IsReconcileRepair(string fullRef, DataValueSnapshot read)
{
if (!_lastValues.TryGetValue(fullRef, out var last)) return true;
if (last.StatusCode == read.StatusCode && Utils.IsEqual(last.Value, read.Value)) return false;
return read.SourceTimestampUtc is not { } src
|| last.SourceTimestampUtc is not { } lastSrc
|| src >= lastSrc;
}
/// <summary>Quality-only annotation (#626 pattern) of every subscribed alarm condition.</summary>
private void AnnotateAlarmQuality(uint quality, DateTime now)
{
@@ -2981,6 +3319,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
catch { /* timeout or already disposed — proceed */ }
_gate.Dispose();
try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { }
SdkTelemetry.Release(_sdkTelemetry);
}
@@ -3010,6 +3349,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
catch { /* timeout or already disposed */ }
_gate.Dispose();
try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { }
SdkTelemetry.Release(_sdkTelemetry);
}
}
@@ -53,11 +53,6 @@
<InputNumber @bind-Value="_form.OpcUa.KeepAliveIntervalSeconds" @bind-Value:after="EmitAsync" class="form-control form-control-sm" />
<div class="form-text">Default 5 s.</div>
</div>
<div class="col-md-3">
<label class="form-label">Reconnect period (s)</label>
<InputNumber @bind-Value="_form.OpcUa.ReconnectPeriodSeconds" @bind-Value:after="EmitAsync" class="form-control form-control-sm" />
<div class="form-text">Initial delay after session drop. Default 5 s.</div>
</div>
<div class="col-md-3">
<label class="form-label">Max discovered nodes</label>
<InputNumber @bind-Value="_form.OpcUa.MaxDiscoveredNodes" @bind-Value:after="EmitAsync" class="form-control form-control-sm" />
@@ -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,
@@ -153,6 +153,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// behaviour for the same reason as <see cref="RediscoveryRaised"/> — a probe tick can land while the
/// driver is between connects.</summary>
private sealed record HostStatusRaised;
/// <summary>Self-sent when the wrapped driver raises <see cref="IDriverConnectionLossSource.ConnectionLost"/> (#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.</summary>
private sealed record ConnectionLossRaised(DriverConnectionLossEventArgs Args);
public sealed class RetryConnect
{
@@ -233,6 +238,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
private EventHandler<AlarmEventArgs>? _alarmEventHandler;
private EventHandler<RediscoveryEventArgs>? _rediscoveryHandler;
private EventHandler<HostStatusChangedEventArgs>? _hostStatusHandler;
private EventHandler<DriverConnectionLossEventArgs>? _connectionLossHandler;
/// <summary>When the driver last raised <see cref="IRediscoverable.OnRediscoveryNeeded"/>, 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<RouteAlarmAck>(_ => { /* stubbed drivers have no alarm backend */ });
Receive<DisconnectObserved>(_ => { /* stubbed drivers don't disconnect */ });
Receive<ConnectionLossRaised>(IgnoreConnectionLoss);
Receive<ForceReconnect>(_ => { /* stubbed drivers don't reconnect */ });
Receive<SetDesiredSubscriptions>(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<SetDesiredSubscriptions>(StoreDesiredSubscriptions);
Receive<ForceReconnect>(_ => { /* already connecting — no-op */ });
Receive<ConnectionLossRaised>(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<SubscriptionEstablished>(msg =>
@@ -500,15 +509,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// reconnects but never manages a read is measured from THIS connect, not from a stale earlier one.
_connectedSinceUtc = DateTime.UtcNow;
Receive<DisconnectObserved>(msg =>
{
_log.Warning("DriverInstance {Id}: disconnect observed ({Reason}); reconnecting",
_driverInstanceId, msg.Reason);
DetachSubscription();
RecordFault();
Become(Reconnecting);
PublishHealthSnapshot();
});
Receive<DisconnectObserved>(msg => HandleDisconnectObserved(msg.Reason));
// #676 — the driver KNOWS its connection is gone (e.g. OpcUaClient: the upstream server halted or forgot the
// session). Same handling as the watchdog's disconnect, minus the ~90 s of silence it would need to prove it.
// Handled inline rather than re-told as DisconnectObserved (Reconnecting also swallows a stray DisconnectObserved).
Receive<ConnectionLossRaised>(msg => HandleDisconnectObserved($"driver reported connection lost: {msg.Args.Reason}"));
Receive<ForceReconnect>(_ =>
{
_log.Info("DriverInstance {Id}: ForceReconnect requested by admin; re-entering Reconnecting", _driverInstanceId);
@@ -764,6 +769,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Receive<InitializeFailed>(_ => { /* keep retrying via timer */ });
Receive<SetDesiredSubscriptions>(StoreDesiredSubscriptions);
Receive<ForceReconnect>(_ => { /* already reconnecting — no-op */ });
Receive<ConnectionLossRaised>(IgnoreConnectionLoss);
// A watchdog DisconnectObserved self-sent on the Connected tick can still be queued when a ConnectionLossRaised
// (or another trip) has already moved us here — already reconnecting, so it is a no-op, not an Unhandled.
Receive<DisconnectObserved>(_ => { /* already reconnecting — no-op */ });
// ResubscribeDesired self-Tells Subscribe; HandleSubscribeAsync replies SubscriptionEstablished to the
// sender, which on the self-resubscribe path is Self. Swallow it (trace only) so it doesn't dead-letter.
Receive<SubscriptionEstablished>(msg =>
@@ -1101,6 +1110,50 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
_rediscoveryHandler = null;
}
/// <summary>
/// Subscribe the driver's <see cref="IDriverConnectionLossSource.ConnectionLost"/> (#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 <c>PreStart</c> and detached in <c>PostStop</c>, like <see cref="AttachRediscoverySource"/>: the
/// <see cref="IDriver"/> instance can outlive this actor, so a per-connect attach would leak a handler (holding
/// a dead <c>Self</c>) per respawn.
/// </summary>
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;
}
/// <summary>Symmetric teardown, called from PostStop — same leak argument as <see cref="DetachRediscoverySource"/>.</summary>
private void DetachConnectionLossSource()
{
if (_driver is IDriverConnectionLossSource src && _connectionLossHandler is not null)
src.ConnectionLost -= _connectionLossHandler;
_connectionLossHandler = null;
}
/// <summary>
/// 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.
/// </summary>
private void IgnoreConnectionLoss(ConnectionLossRaised msg) =>
_log.Debug("DriverInstance {Id}: driver reported connection lost ({Reason}) while not connected — ignored",
_driverInstanceId, msg.Args.Reason);
/// <summary>
/// Connected → Reconnecting on an observed disconnect — from the #527 watchdog or a driver's definitive
/// <see cref="IDriverConnectionLossSource.ConnectionLost"/> (#676).
/// </summary>
private void HandleDisconnectObserved(string reason)
{
_log.Warning("DriverInstance {Id}: disconnect observed ({Reason}); reconnecting", _driverInstanceId, reason);
DetachSubscription();
RecordFault();
Become(Reconnecting);
PublishHealthSnapshot();
}
/// <summary>Subscribe the driver's <see cref="IHostConnectivityProbe.OnHostStatusChanged"/> (if it is
/// one), marshaling each transition to the actor thread. Idempotent; mirrors
/// <see cref="AttachRediscoverySource"/>, including the PreStart-not-per-connect placement — a probe
@@ -1497,6 +1550,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// same driver object), so a missing unsubscribe accumulates a handler per respawn holding a dead Self.
DetachRediscoverySource();
DetachHostStatusSource();
DetachConnectionLossSource();
try { _driver.ShutdownAsync(CancellationToken.None).GetAwaiter().GetResult(); }
catch (Exception ex) { _log.Warning(ex, "DriverInstance {Id}: ShutdownAsync threw on PostStop", _driverInstanceId); }
OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1,
@@ -137,4 +137,17 @@ public sealed class OpcUaClientDriverBrowserTests
sut.LastUsedUtc.ShouldBeGreaterThan(before,
"AttributesAsync must refresh LastUsedUtc to satisfy the IBrowseSession contract");
}
/// <summary>
/// #676: a config still carrying the deleted <c>reconnectPeriod</c> key must deserialise — the browser's own
/// validation (here: Certificate auth refused) is what fires, not a JSON error.
/// </summary>
[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<InvalidOperationException>(
() => _sut.OpenAsync(json, TestContext.Current.CancellationToken));
ex.Message.ShouldContain("Certificate");
}
}
@@ -0,0 +1,662 @@
using System.Text.Json;
using Microsoft.Extensions.Logging;
using Moq;
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary>
/// Gitea #676 — the driver half of the fast recovery path (<see cref="IDriverConnectionLossSource"/>), the
/// per-recovery replay-drop summary, and the trailing reconcile read after an in-place recovery.
/// </summary>
[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<OpcUaClientDriver>
{
public List<(LogLevel Level, string Message)> Entries { get; } = [];
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
lock (Entries) Entries.Add((logLevel, formatter(state, exception)));
}
public string[] Info(string contains) => At(LogLevel.Information, contains);
public string[] At(LogLevel level, string contains)
{
lock (Entries)
return [.. Entries.Where(e => e.Level == level && e.Message.Contains(contains)).Select(e => e.Message)];
}
}
private static RawTagEntry Tag(string rawPath, string nodeId) =>
new(rawPath, JsonSerializer.Serialize(new { nodeId }), WriteIdempotent: true);
private static readonly IReadOnlyList<RawTagEntry> Tags =
[
Tag(SpeedRef, "ns=0;s=Speed"),
Tag(CounterRef, "ns=0;s=Counter"),
];
private static Mock<ISession> NewSession()
{
var mock = new Mock<ISession>(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<ISession> 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);
/// <summary>Every read returns <paramref name="values"/> (one per requested node, in order) at <paramref name="at"/>.</summary>
private static void SetupRead(Mock<ISession> session, Func<DateTime> at, params object[] values) =>
session.Setup(s => s.ReadAsync(
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
.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<uint> DefinitiveStatuses =>
[
StatusCodes.BadSessionIdInvalid,
StatusCodes.BadSessionClosed,
StatusCodes.BadSessionNotActivated,
StatusCodes.BadServerHalted,
];
public static TheoryData<uint> 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<string>();
drv.OnDataChange += (_, e) => order.Add($"bad:{e.Snapshot.StatusCode:X8}");
var raised = new List<DriverConnectionLossEventArgs>();
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<DataChangeEventArgs>();
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<DriverConnectionLossEventArgs>();
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);
}
/// <summary>
/// Review of 3b88056f: only a SERVER-reported subscription loss is definitive. The publishing-stall heuristic
/// (SDK <c>PublishingStopped</c> persisting past a grace) is an inference from silence — it still degrades health
/// and freezes the stamp, but recovery stays with the #527 watchdog.
/// </summary>
[Fact]
public void The_publishing_stall_heuristic_does_not_raise_ConnectionLost()
{
var (drv, _, clock, _) = NewConnectedDriver();
using var _ = drv;
drv.AddSubscriptionForTest([SpeedRef]);
drv.IsPublishingStopped = _ => true;
drv.PublishingStallGrace = _ => TimeSpan.FromSeconds(10);
var raised = 0;
drv.ConnectionLost += (_, _) => raised++;
drv.RaiseKeepAliveForTest(StatusCodes.Good);
clock.Advance(TimeSpan.FromSeconds(11));
drv.RaiseKeepAliveForTest(StatusCodes.Good);
// Positive control: the heuristic DID fire.
drv.GetHealth().LastError.ShouldNotBeNull().ShouldContain("subscription lost");
raised.ShouldBe(0);
}
// ── Replay-drop summary ──────────────────────────────────────────────────────────────────────────────────────
[Fact]
public async Task The_replay_window_logs_one_Information_summary_of_its_drops_when_it_closes()
{
var (drv, session, clock, log) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.FromDays(1); // keep the trailing read out of this test
var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0.AddSeconds(5)));
drv.DeliverNotificationForTest(handle, CounterRef, Good(2, t0.AddSeconds(5)));
SetupRead(session, () => t0.AddSeconds(5), 1, 2);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await drv.RecoveryTaskForTest;
// Three late replays across two refs, one fresh notification that is accepted.
drv.DeliverNotificationForTest(handle, SpeedRef, Good(0, t0));
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0.AddSeconds(5)));
drv.DeliverNotificationForTest(handle, CounterRef, Good(0, t0));
drv.DeliverNotificationForTest(handle, SpeedRef, Good(3, t0.AddSeconds(6)));
log.Info("replay window closed").ShouldBeEmpty("the summary is logged once, when the window closes");
clock.Advance(TimeSpan.FromSeconds(31));
drv.DeliverNotificationForTest(handle, SpeedRef, Good(4, t0.AddSeconds(40))); // first notification after expiry
drv.DeliverNotificationForTest(handle, SpeedRef, Good(5, t0.AddSeconds(41)));
log.Info("replay window closed").ShouldHaveSingleItem()
.ShouldContain("dropped 3 late notifications across 2 refs after recovery");
}
[Fact]
public async Task A_window_with_no_drops_closes_at_Debug_not_Information()
{
var (drv, session, clock, log) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.FromDays(1);
var handle = drv.AddSubscriptionForTest([SpeedRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0));
SetupRead(session, () => t0, 1);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await drv.RecoveryTaskForTest;
clock.Advance(TimeSpan.FromSeconds(31));
drv.DeliverNotificationForTest(handle, SpeedRef, Good(2, t0.AddSeconds(40)));
log.Info("replay window closed").ShouldBeEmpty("nothing was dropped — not worth an Information line per recovery");
log.At(LogLevel.Debug, "replay window closed").ShouldHaveSingleItem().ShouldContain("dropped 0");
}
[Fact]
public async Task A_re_arm_inside_an_open_window_reports_the_earlier_drops_once_and_starts_counting_afresh()
{
var (drv, session, clock, log) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.FromDays(1);
var handle = drv.AddSubscriptionForTest([SpeedRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(1, t0));
SetupRead(session, () => t0, 1);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await drv.RecoveryTaskForTest;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(0, t0.AddSeconds(-1))); // one drop in window 1
clock.Advance(TimeSpan.FromSeconds(5)); // window 1 still open …
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); // … a second outage re-arms it
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await drv.RecoveryTaskForTest;
clock.Advance(TimeSpan.FromSeconds(31));
drv.DeliverNotificationForTest(handle, SpeedRef, Good(3, t0.AddSeconds(80)));
log.Info("replay window closed").ShouldBe(
["OpcUaClientDriver 'opcua-676': post-recovery replay window closed — dropped 1 late notifications across 1 refs after recovery."],
"window 1's drop is reported once, when the re-arm supersedes it — not carried into window 2's summary");
log.At(LogLevel.Debug, "replay window closed").ShouldHaveSingleItem().ShouldContain("dropped 0");
}
// ── Trailing reconcile read ──────────────────────────────────────────────────────────────────────────────────
[Fact]
public async Task The_trailing_re_read_publishes_a_change_the_subscription_lost_and_nothing_it_already_has()
{
var (drv, session, clock, log) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0));
drv.DeliverNotificationForTest(handle, CounterRef, Good(7, t0));
// The recovery read sees both unchanged; by the trailing read Speed has moved on (its notification was lost in
// the stalled publish pipeline) while Counter has not.
var reads = 0;
session.Setup(s => s.ReadAsync(
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
.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<DataChangeEventArgs>();
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<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
.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<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()), Times.Once,
"only the recovery read — the trailing read must not run against a session that is no longer current");
}
/// <summary>Wires the mock session so the first read (the recovery read) returns <paramref name="recovery"/> and
/// every later one (the trailing read) returns <paramref name="trailing"/>.</summary>
private static Func<int> SetupRecoveryThenTrailing(Mock<ISession> session, DataValue recovery, DataValue trailing)
{
var reads = 0;
session.Setup(s => s.ReadAsync(
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(() => new ReadResponse
{
ResponseHeader = new ResponseHeader(),
Results = [Interlocked.Increment(ref reads) > 1 ? trailing : recovery],
DiagnosticInfos = [],
});
return () => Volatile.Read(ref reads);
}
private static async Task<List<(uint Status, object? Value)>> RecoverAndReconcileAsync(
OpcUaClientDriver drv, string fullRef = SpeedRef)
{
var seen = new List<(uint Status, object? Value)>();
drv.OnDataChange += (_, e) => { if (e.FullReference == fullRef) lock (seen) seen.Add((e.Snapshot.StatusCode, e.Snapshot.Value)); };
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await drv.RecoveryTaskForTest;
await drv.ReconcileTaskForTest;
return seen;
}
/// <summary>Review of 3b88056f: a Good→Bad change lost in the stalled pipeline left the node stale-GOOD; the
/// trailing read repairs any (value, status) difference, not just Good ones.</summary>
[Fact]
public async Task The_trailing_re_read_repairs_a_lost_change_to_Bad()
{
var (drv, session, clock, _) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0));
var reads = SetupRecoveryThenTrailing(session, Good(10f, t0),
new DataValue(new Variant(10f), StatusCodes.BadSensorFailure, t0.AddSeconds(9), t0.AddSeconds(9)));
var seen = await RecoverAndReconcileAsync(drv);
reads().ShouldBe(2);
seen.ShouldBe([(BadNoCommunication, (object?)10f), (0u, 10f), ((uint)StatusCodes.BadSensorFailure, 10f)]);
}
/// <summary>A server that sends no source timestamp: the re-read is the server's CURRENT state, read after the
/// stalled pipeline drained, so a differing value is newer than anything delivered before — it is published.</summary>
[Fact]
public async Task The_trailing_re_read_repairs_a_server_that_sends_no_source_timestamp()
{
var (drv, session, clock, _) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
var t0 = clock.Now.UtcDateTime;
DataValue NoSource(object v, DateTime server) =>
new(new Variant(v), StatusCodes.Good, DateTime.MinValue, server);
drv.DeliverNotificationForTest(handle, SpeedRef, NoSource(10f, t0));
var reads = SetupRecoveryThenTrailing(session, NoSource(10f, t0.AddSeconds(1)), NoSource(40f, t0.AddSeconds(9)));
var seen = await RecoverAndReconcileAsync(drv);
reads().ShouldBe(2);
seen.ShouldBe([(BadNoCommunication, (object?)10f), (0u, 10f), (0u, 40f)]);
}
/// <summary>
/// Equal source timestamp, different status: the re-read is the server's current state, not a replay, so it is
/// published — which also repairs the one case the replay guard drops by design (a status-only change that kept
/// its source timestamp inside the window). An IDENTICAL (value, status) is not republished.
/// </summary>
[Fact]
public async Task The_trailing_re_read_publishes_a_status_only_change_at_an_equal_source_time_but_not_an_identical_value()
{
var (drv, session, clock, _) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0));
drv.DeliverNotificationForTest(handle, CounterRef, Good(7, t0));
var reads = 0;
session.Setup(s => s.ReadAsync(
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(() => new ReadResponse
{
ResponseHeader = new ResponseHeader(),
Results = Interlocked.Increment(ref reads) > 1
? [new DataValue(new Variant(10f), StatusCodes.UncertainLastUsableValue, t0, t0), Good(7, t0)]
: [Good(10f, t0), Good(7, t0)],
DiagnosticInfos = [],
});
var seen = new List<(string Ref, uint Status)>();
drv.OnDataChange += (_, e) => { lock (seen) seen.Add((e.FullReference, e.Snapshot.StatusCode)); };
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await drv.RecoveryTaskForTest;
await drv.ReconcileTaskForTest;
reads.ShouldBe(2);
seen.Where(e => e.Status != BadNoCommunication).ShouldBe([
(SpeedRef, 0u), (CounterRef, 0u), // recovery read
(SpeedRef, (uint)StatusCodes.UncertainLastUsableValue), // trailing read: status-only change at t0
]);
}
/// <summary>
/// Review of 3b88056f: the trailing read's compare and its publish are one critical section with the live
/// notification path. Here a NEWER notification arrives exactly between the trailing read's decision and its
/// publish; it must wait, so the older re-read can never land after it and regress the node.
/// </summary>
[Fact]
public async Task A_notification_racing_the_trailing_publish_is_serialised_after_it()
{
var (drv, session, clock, _) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
var t0 = clock.Now.UtcDateTime;
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, t0));
SetupRecoveryThenTrailing(session, Good(10f, t0), Good(40f, t0.AddSeconds(5)));
Task? racer = null;
drv.BeforeReconcilePublishForTest = _ =>
{
racer = Task.Run(() => drv.DeliverNotificationForTest(handle, SpeedRef, Good(50f, t0.AddSeconds(8))));
// Bounded: with the critical section the racer cannot finish until this publish is done.
racer.Wait(TimeSpan.FromMilliseconds(300));
};
var seen = await RecoverAndReconcileAsync(drv);
await racer!.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
lock (seen) seen.Select(e => e.Value).ShouldBe([10f, 10f, 40f, 50f], "the newer 50 must be the last value delivered");
}
/// <summary>Review of 3b88056f: the background timers are bound to the session — a shutdown cancels a pending
/// trailing read instead of leaving it to fire (and read) against whatever session exists later.</summary>
[Fact]
public async Task Shutdown_cancels_a_pending_trailing_read()
{
var (drv, session, clock, _) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.FromHours(1);
var handle = drv.AddSubscriptionForTest([SpeedRef]);
drv.DeliverNotificationForTest(handle, SpeedRef, Good(10f, clock.Now.UtcDateTime));
SetupRead(session, () => clock.Now.UtcDateTime, 10f);
drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication);
drv.RaiseKeepAliveForTest(StatusCodes.Good);
await drv.RecoveryTaskForTest;
drv.ReconcileTaskForTest.IsCompleted.ShouldBeFalse("pending for an hour");
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
await drv.ReconcileTaskForTest.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
}
/// <summary>
/// Re-review of 30a9d5f0: with no source timestamps to compare, "newer" cannot be judged from the values. The
/// trailing read samples V1; while it is in flight a live notification delivers V2; the read then returns V1 —
/// different from the last delivered value and with no timestamp to prove it older. It must NOT be published: a
/// ref delivered to after the read was issued is skipped (per-ref delivery sequence), or the node regresses to V1.
/// </summary>
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task A_notification_delivered_while_the_trailing_read_is_in_flight_is_not_overwritten(bool readHasSourceTime)
{
var (drv, session, clock, _) = NewConnectedDriver();
using var _ = drv;
drv.PostRecoveryReconcileDelay = () => TimeSpan.Zero;
var handle = drv.AddSubscriptionForTest([SpeedRef]);
var t0 = clock.Now.UtcDateTime;
DataValue NoSource(object v, DateTime server) => new(new Variant(v), StatusCodes.Good, DateTime.MinValue, server);
drv.DeliverNotificationForTest(handle, SpeedRef, NoSource(10f, t0));
var reads = 0;
session.Setup(s => s.ReadAsync(
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<TimestampsToReturn>(),
It.IsAny<ReadValueIdCollection>(), It.IsAny<CancellationToken>()))
.ReturnsAsync(() =>
{
DataValue result;
if (Interlocked.Increment(ref reads) == 1)
result = NoSource(10f, t0.AddSeconds(1)); // recovery read: unchanged
else
{
// The trailing read has sampled V1 at the server; before its response is processed, the live
// subscription delivers the newer V2.
drv.DeliverNotificationForTest(handle, SpeedRef, NoSource(20f, t0.AddSeconds(9)));
result = readHasSourceTime
? Good(15f, t0.AddSeconds(8)) // read carries a source time, V2 did not
: NoSource(15f, t0.AddSeconds(8));
}
return new ReadResponse { ResponseHeader = new ResponseHeader(), Results = [result], DiagnosticInfos = [] };
});
var seen = await RecoverAndReconcileAsync(drv);
reads.ShouldBe(2, "the trailing read must have run");
lock (seen) seen.Select(e => e.Value).ShouldBe([10f, 10f, 20f], "the read's stale 15 must not overwrite the newer 20");
}
}
@@ -75,4 +75,23 @@ public class OpcUaClientDriverFactoryTests
opts.RawTags[0].WriteIdempotent.ShouldBeFalse();
opts.RawTags[1].WriteIdempotent.ShouldBeTrue();
}
/// <summary>
/// #676 deleted the dead <c>ReconnectPeriod</c> 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.
/// </summary>
[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");
}
}
@@ -134,4 +134,27 @@ public sealed class OpcUaClientDriverProbeTests
await acceptTask.WaitAsync(TimeSpan.FromSeconds(5), testCt);
}
}
/// <summary>
/// #676: a stored config still carrying the deleted <c>reconnectPeriod</c> key must reach the connect phase, not
/// fail as invalid JSON — the probe parses with the same skip-unknown-members options as the factory.
/// </summary>
[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);
}
}
@@ -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));
}
/// <summary>Verifies that driver reports type and ID before connecting.</summary>
@@ -0,0 +1,69 @@
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary>
/// 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
/// <c>CoreClientUtils.SelectEndpoint</c> does.
/// </summary>
public sealed class OpcUaClientEndpointRewriteTests
{
[Fact]
public void Host_and_port_of_the_advertised_url_are_replaced_by_the_configured_ones()
{
OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://internal-host:4840/OtOpcUa", "opc.tcp://10.1.2.3:54840/OtOpcUa")
.ShouldBe("opc.tcp://10.1.2.3:54840/OtOpcUa");
}
[Fact]
public void The_advertised_path_is_kept()
{
// The path is the server's to choose (several endpoints may share a host:port); only the address is ours.
OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://plc-a:4840/Server/Ua", "opc.tcp://gateway:6000/")
.ShouldBe("opc.tcp://gateway:6000/Server/Ua");
}
[Fact]
public void Only_the_port_differs_still_rewrites()
{
// The in-process stall proxy shape: same host, the configured port is the proxy's.
OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://127.0.0.1:4840/OtOpcUa", "opc.tcp://127.0.0.1:61000/OtOpcUa")
.ShouldBe("opc.tcp://127.0.0.1:61000/OtOpcUa");
}
[Fact]
public void A_configured_url_without_a_port_keeps_the_advertised_port()
{
// Review of 3b88056f: Uri.Port is -1 for a port-less opc.tcp URL; copying it would drop the advertised port and
// the SDK would fall back to 4840.
OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://internal-host:62541/Server", "opc.tcp://gateway/Server")
.ShouldBe("opc.tcp://gateway:62541/Server");
}
[Fact]
public void An_IPv6_configured_host_is_bracketed()
{
OpcUaClientDriver.RewriteEndpointUrl("opc.tcp://internal-host:4840/OtOpcUa", "opc.tcp://[fd00::1]:54840/OtOpcUa")
.ShouldBe("opc.tcp://[fd00::1]:54840/OtOpcUa");
}
[Fact]
public void A_different_scheme_is_left_alone()
{
// An https endpoint advertised next to an opc.tcp one is a different transport, not a NAT artefact.
OpcUaClientDriver.RewriteEndpointUrl("https://srv:443/ua", "opc.tcp://srv:4840/ua")
.ShouldBe("https://srv:443/ua");
}
[Theory]
[InlineData(null)]
[InlineData("")]
[InlineData("not a url")]
public void An_unusable_advertised_url_falls_back_to_the_configured_one(string? advertised)
{
OpcUaClientDriver.RewriteEndpointUrl(advertised, "opc.tcp://srv:4840/ua").ShouldBe("opc.tcp://srv:4840/ua");
}
}
@@ -4,29 +4,15 @@ using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
/// <summary>
/// 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);
/// <c>ReconnectPeriod</c> is retained for config compatibility only. Outage behaviour is covered by
/// <see cref="OpcUaClientSessionLossTests"/> and, over the wire, <c>OpcUaClientDriverReconnectWireTests</c>.
/// 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 <c>ReconnectPeriod</c> option; <see cref="OpcUaClientDriverFactoryTests"/> pins that
/// stored configs still carrying it parse. Outage behaviour is covered by <see cref="OpcUaClientSessionLossTests"/>,
/// <see cref="OpcUaClientConnectionLossSignalTests"/> and, over the wire, <c>OpcUaClientDriverReconnectWireTests</c>.
/// </summary>
[Trait("Category", "Unit")]
public sealed class OpcUaClientReconnectTests
{
/// <summary>Verifies that the default reconnect period matches the driver specification of 5 seconds.</summary>
[Fact]
public void Default_ReconnectPeriod_matches_driver_specs_5_seconds()
{
new OpcUaClientDriverOptions().ReconnectPeriod.ShouldBe(TimeSpan.FromSeconds(5));
}
/// <summary>Verifies that reconnect period can be configured for aggressive or relaxed retries.</summary>
[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));
}
/// <summary>Verifies that the driver starts in the Unknown state before initialization.</summary>
[Fact]
public void Driver_starts_unknown_pre_init()
@@ -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);
@@ -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 <c>DriverInstanceActor</c>
/// 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 (<c>BadServerHalted</c>), so the driver also raises
/// <c>ConnectionLost</c> and the host runs that sequence at once instead of after the watchdog's silence window.
/// </summary>
[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<string>();
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");
}
/// <summary>
@@ -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<string>();
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);
@@ -0,0 +1,159 @@
using Opc.Ua;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>
/// 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 (<c>docker pause</c> of opc-plc) covered it. <see cref="TcpStallProxy"/> 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.
/// </summary>
/// <remarks>
/// 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
/// (<c>ReconcileAfterRecoveryAsync</c>) is the fix; the "40.0 arrives" leg below goes red without it.
/// </remarks>
[Trait("Category", "Integration")]
public sealed class OpcUaClientDriverStallWireTests
{
private const uint BadNoCommunication = 0x80310000u;
private static async Task WaitAsync(Func<bool> condition, string because, TimeSpan? timeout = null)
{
var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(30));
while (DateTime.UtcNow < deadline)
{
if (condition()) return;
await Task.Delay(25);
}
condition().ShouldBeTrue(because);
}
private sealed record Seen(uint Status, object? Value, DateTime? Source);
[Fact]
public async Task A_stalled_connection_goes_Bad_once_and_recovers_in_place_on_the_same_session_without_regressing()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-stall", ct);
await using var proxy = TcpStallProxy.Start(new Uri(rig.EndpointUrl).Port);
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
// A 3 s operation timeout keeps the stall short (a keep-alive is declared failed when its request times out); a
// 120 s session timeout and a 3000-cycle (300 s at 100 ms) subscription lifetime keep both far outside the
// stall, so the test exercises the in-place recovery and nothing else.
await using var drv = OpcUaClientDriverWireTests.NewDriver(
rig, "wire-stall", endpointUrl: proxy.Through(rig.EndpointUrl), timeout: TimeSpan.FromSeconds(3),
sessionTimeout: TimeSpan.FromSeconds(120), subscriptionLifetimeCount: 3000);
var connectionLost = 0;
drv.ConnectionLost += (_, _) => Interlocked.Increment(ref connectionLost);
await drv.InitializeAsync("{}", ct);
proxy.AcceptedConnections.ShouldBeGreaterThan(0, "the session must run THROUGH the proxy, or pausing it proves nothing");
var seen = new List<Seen>();
drv.OnDataChange += (_, e) =>
{
if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef)
lock (seen) seen.Add(new Seen(e.Snapshot.StatusCode, e.Snapshot.Value, e.Snapshot.SourceTimestampUtc));
};
await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct);
await WaitAsync(() => { lock (seen) return seen.Any(s => s is { Status: 0u, Value: 10.0f }); }, "initial value");
var session = drv.Session.ShouldNotBeNull();
var connectionsBefore = proxy.AcceptedConnections;
// Bad keep-alive TICKS: the SDK reports a failed tick as two events at the same instant (the request's status and
// BadNoCommunication), so events closer than half a keep-alive interval are one tick.
var badKeepAlives = 0;
var lastBadTick = DateTime.MinValue;
session.KeepAlive += (_, e) =>
{
if (!ServiceResult.IsBad(e.Status)) return;
lock (seen)
{
if (DateTime.UtcNow - lastBadTick < TimeSpan.FromMilliseconds(500)) return;
lastBadTick = DateTime.UtcNow;
badKeepAlives++;
}
};
// The SDK's late backlog flush: a data-carrying NotificationMessage whose sequence number is BELOW one already
// seen (keep-alive messages carry the next number). It is the positive control for the no-regression check.
uint maxSequence = 0;
var lateMessages = 0;
session.Notification += (_, e) =>
{
var msg = e.NotificationMessage;
if (msg is null) return;
lock (seen)
{
if (msg.SequenceNumber < maxSequence && msg.NotificationData.Count > 0) lateMessages++;
maxSequence = Math.Max(maxSequence, msg.SequenceNumber);
}
};
// ── Stall ────────────────────────────────────────────────────────────────────────────────────────────────
proxy.Pause();
await WaitAsync(() => { lock (seen) return seen.Any(s => s.Status == BadNoCommunication); },
"a stalled upstream must publish BadNoCommunication for the subscribed ref");
lock (seen) seen.Single(s => s.Status == BadNoCommunication).Value.ShouldBe(10.0f, "the Bad retains the value");
drv.GetHealth().State.ShouldNotBe(DriverState.Healthy);
var frozen = drv.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
// The server's value moves on while the client cannot see it: away and back to the pre-outage value, so the
// recovered value is IDENTICAL to the one published Bad — the (value, status) dedup trap.
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 20.0f);
await Task.Delay(300, ct); // > one publishing interval, so the server queues 20.0 as its own notification
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
// Stays Bad and frozen across one further failed keep-alive tick (counted, not slept for).
int badAtDetection;
lock (seen) badAtDetection = badKeepAlives;
await WaitAsync(() => { lock (seen) return badKeepAlives > badAtDetection; }, "a further keep-alive must fail while stalled");
drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen, "the read stamp is the #527 watchdog's silence measure");
lock (seen) seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1, "one Bad per outage, not per keep-alive");
// ── Resume well inside the 30 s session timeout ──────────────────────────────────────────────────────────
proxy.Resume();
await WaitAsync(() => drv.GetHealth().State == DriverState.Healthy, "a resumed stall must recover in place");
int badIndex;
lock (seen) badIndex = seen.FindIndex(s => s.Status == BadNoCommunication);
await WaitAsync(() => { lock (seen) return seen.Skip(badIndex + 1).Any(s => s is { Status: 0u, Value: 10.0f }); },
"the identical value must be republished Good after the outage");
drv.Session.ShouldBeSameAs(session, "an in-place blip must not cost the session");
proxy.AcceptedConnections.ShouldBe(connectionsBefore, "and not the TCP connection either");
// A change published straight after recovery — while the Publish requests held across the stall are still
// being answered or timed out — must still reach the node.
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 40.0f);
await WaitAsync(() => { lock (seen) return seen.Any(s => s is { Status: 0u, Value: 40.0f }); },
"a change published right after recovery must reach the node (via the subscription or the trailing re-read)",
TimeSpan.FromSeconds(60));
await drv.ReconcileTaskForTest;
// The SDK flushes the notifications held during the stall seconds after recovery (the #670 F2 shape). Wait for
// that flush before judging ordering, or the check below would pass on a run where nothing late arrived.
await WaitAsync(() => { lock (seen) return lateMessages > 0; },
"the SDK's late backlog flush did not arrive — the no-regression check below would be vacuous",
TimeSpan.FromSeconds(60));
lock (seen)
{
var good = seen.Where(s => s.Status == 0u && s.Source is not null).ToList();
for (var i = 1; i < good.Count; i++)
good[i].Source!.Value.ShouldBeGreaterThanOrEqualTo(good[i - 1].Source!.Value,
$"published Good value #{i} ({good[i].Value}) regressed in source time — a late replay reached the " +
$"node. Sequence: [{string.Join(", ", good.Select(s => $"{s.Value}@{s.Source:HH:mm:ss.fff}"))}]");
seen.Last().Value.ShouldBe(40.0f, "nothing older may land after the latest value");
seen.Count(s => s.Status == BadNoCommunication).ShouldBe(1);
}
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
drv.Session.ShouldBeSameAs(session);
Volatile.Read(ref connectionLost).ShouldBe(0,
"a stall is AMBIGUOUS (the session survived it) — it must go through the #527 watchdog, never the fast path");
}
}
@@ -41,10 +41,11 @@ public sealed class OpcUaClientDriverWireTests
internal static OpcUaClientDriver NewDriver(
ServerEventRig rig, string id, string? browseRoot = null, IReadOnlyList<RawTagEntry>? extraTags = null,
TimeSpan? reconnectPeriod = null) =>
string? endpointUrl = null, TimeSpan? timeout = null, TimeSpan? sessionTimeout = null,
uint? subscriptionLifetimeCount = null) =>
new(new OpcUaClientDriverOptions
{
EndpointUrl = rig.EndpointUrl,
EndpointUrl = endpointUrl ?? rig.EndpointUrl,
RawTags =
[
Tag(SpeedRef, ServerEventRig.RawNsu(SpeedId)),
@@ -58,11 +59,11 @@ public sealed class OpcUaClientDriverWireTests
Username = "apikey",
Password = ServerEventRig.KeyPassword,
AutoAcceptCertificates = true,
Timeout = TimeSpan.FromSeconds(10),
SessionTimeout = TimeSpan.FromSeconds(30),
Timeout = timeout ?? TimeSpan.FromSeconds(10),
SessionTimeout = sessionTimeout ?? TimeSpan.FromSeconds(30),
SubscriptionLifetimeCount = subscriptionLifetimeCount ?? 1000,
KeepAliveInterval = TimeSpan.FromSeconds(1),
BrowseRoot = browseRoot,
ReconnectPeriod = reconnectPeriod ?? TimeSpan.FromSeconds(5),
}, id);
internal static async Task<ServerEventRig> StartRigAsync(string name, CancellationToken ct)
@@ -0,0 +1,126 @@
using System.Net;
using System.Net.Sockets;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>
/// Gitea #676 — an in-test TCP forwarder that can STALL a connection without closing it: the shape of a
/// <c>docker pause</c>, 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 <see cref="Pause"/> holds all traffic in both directions (bytes already read
/// are held, the rest back up in the kernel buffers) and <see cref="Resume"/> 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.
/// </summary>
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<TcpClient> _sockets = [];
private readonly Lock _lock = new();
private TaskCompletionSource _gate = Open();
private Task _acceptLoop = Task.CompletedTask;
private TcpStallProxy(int targetPort) => _targetPort = targetPort;
/// <summary>The loopback port clients connect to.</summary>
public int Port => ((IPEndPoint)_listener.LocalEndpoint).Port;
/// <summary>Number of client connections accepted so far — a new one means the client re-connected.</summary>
public int AcceptedConnections { get; private set; }
/// <summary>Starts forwarding <c>127.0.0.1:{Port}</c> to <c>127.0.0.1:{targetPort}</c>.</summary>
public static TcpStallProxy Start(int targetPort)
{
var proxy = new TcpStallProxy(targetPort);
proxy._listener.Start();
proxy._acceptLoop = proxy.AcceptLoopAsync();
return proxy;
}
/// <summary>Rewrites the port of an <c>opc.tcp</c> URL to this proxy's.</summary>
public string Through(string endpointUrl) => new UriBuilder(endpointUrl) { Port = Port }.Uri.ToString();
/// <summary>Stops all traffic in both directions; sockets stay open.</summary>
public void Pause()
{
lock (_lock)
if (_gate.Task.IsCompleted) _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
}
/// <summary>Releases everything held since <see cref="Pause"/>, in order.</summary>
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();
}
}
@@ -0,0 +1,175 @@
using Akka.Actor;
using Akka.Event;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// Gitea #676 — a driver that KNOWS its connection is gone (<see cref="IDriverConnectionLossSource"/>) 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.
/// </summary>
public sealed class DriverInstanceActorConnectionLossSignalTests : RuntimeActorTestBase
{
private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(50);
/// <summary>A permanently Healthy (fresh-read) stub — the watchdog can never fire for it — that can raise
/// <see cref="IDriverConnectionLossSource.ConnectionLost"/> on demand and counts its attached handlers.</summary>
private sealed class LossSourceStubDriver : StubDriver, IDriverConnectionLossSource
{
private EventHandler<DriverConnectionLossEventArgs>? _handlers;
private int _handlerCount;
private int _healthReads;
public int HandlerCount => Volatile.Read(ref _handlerCount);
/// <summary>Health-poll observations — the positive control for an absence assertion.</summary>
public int HealthReadCount => Volatile.Read(ref _healthReads);
public override DriverHealth GetHealth()
{
Interlocked.Increment(ref _healthReads);
return base.GetHealth();
}
public event EventHandler<DriverConnectionLossEventArgs>? ConnectionLost
{
add { _handlers += value; Interlocked.Increment(ref _handlerCount); }
remove { _handlers -= value; Interlocked.Decrement(ref _handlerCount); }
}
/// <summary>Raised from a pool thread, as a real driver's keep-alive callback would.</summary>
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<DriverInstanceActor.ConnectivityChanged>(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<DriverInstanceActor.ConnectivityChanged>(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");
}
/// <summary>
/// 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.
/// </summary>
[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");
}
/// <summary>
/// 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.
/// </summary>
[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<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3)).Connected.ShouldBeTrue();
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(600));
driver.InitializeCount.ShouldBe(1);
}
/// <summary>Review of 3b88056f: a watchdog DisconnectObserved still queued when the connection-loss signal has
/// already moved the actor to Reconnecting is a no-op there, not an Unhandled message.</summary>
[Fact]
public async Task A_stray_DisconnectObserved_while_reconnecting_is_not_unhandled()
{
var driver = new LossSourceStubDriver();
var (actor, parent) = Connect(driver);
var unhandled = CreateTestProbe();
Sys.EventStream.Subscribe(unhandled.Ref, typeof(UnhandledMessage));
await driver.RaiseAsync("keep-alive reported BadServerHalted");
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3)).Connected.ShouldBeFalse();
actor.Tell(new DriverInstanceActor.DisconnectObserved("late watchdog trip"));
unhandled.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
}
[Fact]
public void The_handler_is_attached_once_and_detached_when_the_actor_stops()
{
var driver = new LossSourceStubDriver();
var (actor, parent) = Connect(driver);
driver.HandlerCount.ShouldBe(1);
parent.Watch(actor);
Sys.Stop(actor);
// PostStop announces comms loss (03/S15) before the Terminated arrives.
parent.FishForMessage(m => m is Terminated t && t.ActorRef.Equals(actor), TimeSpan.FromSeconds(3));
driver.HandlerCount.ShouldBe(0, "the IDriver outlives the actor on a respawn — a leaked handler holds a dead Self");
}
/// <summary>A driver that does not implement the interface is simply never subscribed — the signal is opt-in.</summary>
[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<DriverInstanceActor.ConnectivityChanged>(TimeSpan.FromSeconds(3)).Connected.ShouldBeTrue();
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
}