dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(opcuaclient): lifecycle generation, close the local session on init throw, one teardown budget (R06:O-2, O-9)

Browse Source 
- Every InitializeAsync/ShutdownAsync bumps a lifecycle generation. The session,
  namespace map and endpoint are installed under _probeLock only while the init's
  generation is current, so an init the host abandoned that finishes after the next
  shutdown discards its session instead of resurrecting it. Every health write
  (lifecycle, keep-alive, publish-error, read path) is generation-stamped, so a late
  writer can no longer overwrite newer state.
- O-2: the init catch closes and disposes the LOCAL session it opened (it closed the
  still-null Session property, orphaning an activated session + keep-alive timer).
- O-9: ShutdownAsync spends ONE budget (4 s, inside the host's 5 s PostStop budget) on
  wire calls: every subscription delete concurrently, then the session close with
  what remains; waits are abandoned when the budget fires even if the SDK ignores the
  token. It used to be 5 s per subscription, serially.

Refs #743

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 15:19:48 -04:00
parent 4d0d5d229d
commit 6c3a754be9
2 changed files with 444 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
+224 -47
View File
@@ -449,11 +449,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <inheritdoc />
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// Every lifecycle transition (initialise / shutdown) takes a new generation. Everything this attempt
// publishes — the session, the namespace map, health — is published only while its generation is still
// the current one, so an init the host abandoned (30 s bound) that finishes after the next ShutdownAsync
// or InitializeAsync cannot resurrect a session, or overwrite the newer health, behind its back.
var generation = Interlocked.Increment(ref _lifecycleGeneration);
// InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract
// allows it, and a reconnect retry can overlap a running attempt); `Session = session` below would otherwise orphan the previous session,
// its keep-alive handler and any reconnect handler.
await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false);
_health = new DriverHealth(DriverState.Initializing, null, null);
await ReleaseSessionAsync(generation, cancellationToken).ConfigureAwait(false);
WriteLifecycleHealth(generation, _ => new DriverHealth(DriverState.Initializing, null, null));
ISession? session = null;
KeepAliveEventHandler? keepAliveHandler = null;
var installed = false;
try
{
// re-parse the supplied DriverConfig so a config change delivered through the IDriver
@@ -512,15 +521,16 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// aggregate exception names every URL that was tried and why — critical diag for
// operators debugging 'why did the failover pick #3?'.
var attemptErrors = new List<string>(candidates.Count);
ISession? session = null;
string? connectedUrl = null;
foreach (var url in candidates)
{
try
{
session = await OpenSessionOnEndpointAsync(
appConfig, url, _options.SecurityPolicy, _options.SecurityMode,
identity, cancellationToken).ConfigureAwait(false);
session = OpenSessionForTest is { } open
? await open(url, cancellationToken).ConfigureAwait(false)
: await OpenSessionOnEndpointAsync(
appConfig, url, _options.SecurityPolicy, _options.SecurityMode,
identity, cancellationToken).ConfigureAwait(false);
connectedUrl = url;
break;
}
@@ -541,8 +551,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// liveness evidence (advances LastSuccessfulRead); a bad one degrades health and publishes the
// outage. Recovery from a session that does not come back is the host's comms-loss watchdog
// (ShutdownAsync + InitializeAsync + re-subscribe) — there is no in-driver reconnect handler.
_keepAliveHandler = OnKeepAlive;
session.KeepAlive += _keepAliveHandler;
// The handler is attached to THIS session only; OnKeepAlive ignores any sender that is not the
// installed session, so a session discarded below never reaches health.
keepAliveHandler = OnKeepAlive;
session.KeepAlive += keepAliveHandler;
AttachDiagnostics(session);
// Build the bidirectional namespace map from the freshly negotiated session's
@@ -551,23 +563,98 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// reorder across a restart can't silently misaddress nodes. v3: the same map also
// ingests the authored RawTags so read/write/subscribe/history references (RawPaths)
// resolve to their upstream node ids.
_namespaceMap = NamespaceMap.FromSession(session, _options.RawTags);
RecordSkippedTags(_namespaceMap);
var map = NamespaceMap.FromSession(session, _options.RawTags);
Session = session;
_connectedEndpointUrl = connectedUrl;
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
TransitionTo(HostState.Running);
// Publish the session only while this attempt is still the current lifecycle generation. The
// generation check and the install share _probeLock with ReleaseSessionAsync's take, so a
// ShutdownAsync either sees (and releases) this session or makes this attempt discard it.
lock (_probeLock)
{
if (Volatile.Read(ref _lifecycleGeneration) == generation)
{
_namespaceMap = map;
Session = session;
_keepAliveHandler = keepAliveHandler;
_connectedEndpointUrl = connectedUrl;
installed = true;
}
}
if (!installed)
throw new OperationCanceledException(
$"OpcUaClientDriver '{_driverInstanceId}': initialise was superseded by a later lifecycle " +
"transition (shutdown or re-initialise); its session was discarded.");
RecordSkippedTags(map);
WriteLifecycleHealth(generation, _ => new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
if (Volatile.Read(ref _lifecycleGeneration) == generation) TransitionTo(HostState.Running);
}
catch (Exception ex)
{
try { if (Session is Session s) await s.CloseAsync().ConfigureAwait(false); } catch { }
Session = null;
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
// R06:O-2 — close the LOCAL session this attempt opened (the property is still null, or belongs to a
// newer attempt), so a throw after the session activated (AttachDiagnostics, the namespace map, the
// superseded check) does not orphan it and its keep-alive timer until GC.
if (session is not null && !installed)
await DiscardSessionAsync(session, keepAliveHandler).ConfigureAwait(false);
if (Volatile.Read(ref _lifecycleGeneration) == generation)
{
if (installed)
await ReleaseSessionAsync(generation, CancellationToken.None).ConfigureAwait(false);
WriteLifecycleHealth(generation,
h => new DriverHealth(DriverState.Faulted, h.LastSuccessfulRead, ex.Message));
}
throw;
}
}
/// <summary>
/// Test seam: opens the session for one endpoint URL in place of the real discovery + session create, so
/// the post-connect half of <see cref="InitializeAsync"/> (and its superseded / failure paths) can be driven
/// against a mock session. Production leaves it null.
/// </summary>
internal Func<string, CancellationToken, Task<ISession>>? OpenSessionForTest { get; set; }
/// <summary>
/// Close and dispose a session this driver never installed (a failed or superseded initialise): detach the
/// handlers attached to it, close it within <see cref="SessionCloseTimeout"/>, dispose it. Best-effort.
/// </summary>
/// <param name="session">The session to discard.</param>
/// <param name="keepAliveHandler">The keep-alive handler attached to it, if any.</param>
private async Task DiscardSessionAsync(ISession session, KeepAliveEventHandler? keepAliveHandler)
{
if (keepAliveHandler is not null)
{
try { session.KeepAlive -= keepAliveHandler; } catch { /* best-effort */ }
}
DetachDiagnostics(session);
try
{
using var closeCts = new CancellationTokenSource(SessionCloseTimeout);
await CloseBoundedAsync(session, closeCts.Token).ConfigureAwait(false);
}
catch { /* best-effort */ }
try { session.Dispose(); } catch { /* best-effort */ }
}
/// <summary>
/// Close a session, abandoning the wait when <paramref name="ct"/> fires even if the SDK call itself ignores
/// the token (a half-dead upstream must not hold a teardown past its budget).
/// </summary>
private static async Task CloseBoundedAsync(ISession session, CancellationToken ct)
{
var close = session.CloseAsync((int)SessionCloseTimeout.TotalMilliseconds, true, ct) ?? Task.FromResult<StatusCode>(StatusCodes.Good);
await close.WaitAsync(ct).ConfigureAwait(false);
}
/// <summary>
/// Write a lifecycle health snapshot only while <paramref name="generation"/> is still the current lifecycle
/// generation (CAS against concurrent keep-alive / data-path writers, re-checking the generation each try).
/// </summary>
/// <param name="generation">The generation of the init / shutdown making the write.</param>
/// <param name="next">Maps the current health to the one to write.</param>
private void WriteLifecycleHealth(long generation, Func<DriverHealth, DriverHealth> next) =>
UpdateHealth(generation, next);
/// <summary>
/// Build a minimal in-memory <see cref="ApplicationConfiguration"/>. Certificates live
/// under the OS user profile — on Windows that's <c>%LocalAppData%\OtOpcUa\pki</c>
@@ -1050,15 +1137,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// Detaches the keep-alive handler and closes + disposes the current session. The session part of
/// <see cref="ShutdownAsync"/>; also run at the top of <see cref="InitializeAsync"/> so a re-initialise on a
/// live driver does not orphan the previous session. Subscriptions are NOT touched here. Clears the
/// outage latch, so the next session reports its own loss.
/// outage latch, so the next session reports its own loss. Does nothing when <paramref name="generation"/> is
/// no longer the current lifecycle generation — the session installed then belongs to a NEWER init, which a
/// late (abandoned) shutdown must not close.
/// </summary>
private async Task ReleaseSessionAsync(CancellationToken cancellationToken)
/// <param name="generation">The lifecycle generation of the caller.</param>
/// <param name="cancellationToken">Bounds the close (on top of <see cref="SessionCloseTimeout"/>).</param>
private async Task ReleaseSessionAsync(long generation, CancellationToken cancellationToken)
{
// Take the session reference under _probeLock and clear Session to null so any concurrent caller that
// checks inside _gate sees null immediately after release begins.
ISession? sessionToClose;
lock (_probeLock)
{
if (Volatile.Read(ref _lifecycleGeneration) != generation) return;
sessionToClose = Session;
if (_keepAliveHandler is not null && sessionToClose is not null)
{
@@ -1075,11 +1167,11 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
var sessionKnownLost = Volatile.Read(ref _sessionLost) == 1;
try
{
if (sessionToClose is Session s && !sessionKnownLost)
if (sessionToClose is not null && !sessionKnownLost)
{
using var closeCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
closeCts.CancelAfter(SessionCloseTimeout);
await s.CloseAsync(closeCts.Token).ConfigureAwait(false);
await CloseBoundedAsync(sessionToClose, closeCts.Token).ConfigureAwait(false);
}
}
catch { /* best-effort */ }
@@ -1107,9 +1199,29 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>Upper bound on the best-effort wire calls a teardown makes against a possibly-dead upstream.</summary>
private static readonly TimeSpan SessionCloseTimeout = TimeSpan.FromSeconds(5);
/// <summary>
/// R06:O-9 — the ONE budget the whole of <see cref="ShutdownAsync"/> spends on wire calls (every subscription
/// delete, concurrently, then the session close). It used to be 5 s per subscription, serially, so a
/// half-dead upstream held the actor's PostStop for N×5 s. Under the host's 5 s PostStop budget for
/// shutdown + dispose together, so the local teardown still runs inside it. Test seam.
/// </summary>
internal TimeSpan TeardownBudget { get; set; } = TimeSpan.FromSeconds(4);
/// <summary>Test seam: the wire delete of one subscription. Production calls the SDK's silent delete.</summary>
internal Func<Subscription, CancellationToken, Task> DeleteSubscriptionAsync { get; set; } =
static (subscription, ct) => subscription.DeleteAsync(silent: true, ct);
/// <inheritdoc />
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
// A new lifecycle generation: an init still running (abandoned by the host) can no longer install its
// session or write health, and keep-alive / data-path health writes of the old session stop landing.
var generation = Interlocked.Increment(ref _lifecycleGeneration);
// R06:O-9 — one budget for every wire call below, linked to the caller's token.
using var budget = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
budget.CancelAfter(TeardownBudget);
// Tear down remote subscriptions first — otherwise Session.Close will try and may fail
// with BadSubscriptionIdInvalid noise in the upstream log. _subscriptions is cleared
// whether or not the wire-side delete succeeds since the local handles are useless
@@ -1119,20 +1231,23 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// against a session already known lost (bad keep-alive) the wire deletes cannot succeed and would each
// sit out the operation timeout — skip them; disposing the session releases the local objects.
var sessionAlive = Volatile.Read(ref _sessionLost) == 0;
foreach (var rs in _subscriptions.Values)
var deletes = new List<Task>();
foreach (var key in _subscriptions.Keys)
{
if (!_subscriptions.TryRemove(key, out var rs)) continue;
DetachNotificationHandlers(rs.ItemHandlers);
if (sessionAlive) await DeleteSubscriptionBestEffortAsync(rs.Subscription, cancellationToken).ConfigureAwait(false);
if (sessionAlive) deletes.Add(DeleteSubscriptionBestEffortAsync(rs.Subscription, budget.Token));
}
_subscriptions.Clear();
foreach (var ras in _alarmSubscriptions.Values)
foreach (var key in _alarmSubscriptions.Keys)
{
if (!_alarmSubscriptions.TryRemove(key, out var ras)) continue;
try { ras.EventItem.Notification -= ras.Handler; }
catch { /* best-effort */ }
if (sessionAlive) await DeleteSubscriptionBestEffortAsync(ras.Subscription, cancellationToken).ConfigureAwait(false);
if (sessionAlive) deletes.Add(DeleteSubscriptionBestEffortAsync(ras.Subscription, budget.Token));
}
_alarmSubscriptions.Clear();
// Concurrently, under the one budget (each delete already swallows its own failure).
await Task.WhenAll(deletes).ConfigureAwait(false);
// The host re-subscribes after a reinit and the new subscriptions deliver initial values; nothing of the
// previous connection's value bookkeeping applies to them.
_lastValues.Clear();
@@ -1151,9 +1266,17 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// ConditionRefresh re-raise every active+acked alarm (an un-ack flicker). BeginAlarmSubscription forgets
// them when the set changes, which is when the host did rebuild.
await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false);
_namespaceMap = null;
_connectedEndpointUrl = null;
// The session close spends what is left of the same budget (none left ⇒ dispose without closing).
await ReleaseSessionAsync(generation, budget.Token).ConfigureAwait(false);
lock (_probeLock)
{
// A newer initialise may have installed its own map / endpoint meanwhile — leave those alone.
if (Volatile.Read(ref _lifecycleGeneration) == generation)
{
_namespaceMap = null;
_connectedEndpointUrl = null;
}
}
// Detach the cert-validation handler so the (potentially process-shared)
// CertificateValidator doesn't hold a delegate to a shutting-down driver.
@@ -1164,18 +1287,19 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_certValidatorRef = null;
}
TransitionTo(HostState.Unknown);
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
if (Volatile.Read(ref _lifecycleGeneration) == generation) TransitionTo(HostState.Unknown);
WriteLifecycleHealth(generation, h => new DriverHealth(DriverState.Unknown, h.LastSuccessfulRead, null));
}
/// <summary>Best-effort, bounded delete of one subscription (shutdown path).</summary>
private static async Task DeleteSubscriptionBestEffortAsync(Subscription subscription, CancellationToken cancellationToken)
/// <summary>
/// Best-effort delete of one subscription (shutdown path), bounded by the shutdown's budget: the wait is
/// abandoned when the budget fires even if the SDK call ignores the token.
/// </summary>
private async Task DeleteSubscriptionBestEffortAsync(Subscription subscription, CancellationToken budget)
{
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(SessionCloseTimeout);
await subscription.DeleteAsync(silent: true, cts.Token).ConfigureAwait(false);
await DeleteSubscriptionAsync(subscription, budget).WaitAsync(budget).ConfigureAwait(false);
}
catch { /* best-effort */ }
}
@@ -1216,7 +1340,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// Re-read Session inside the critical section: if a reconnect completed while we
// were blocked on _gate, a re-initialise has already swapped in the new
// session. NodeId parsing is namespace-relative, so it must also use the current
// session's namespace table.
// session's namespace table. The generation is read first: a shutdown / re-initialise that releases this
// session while the read is in flight bumps it, so the read's health outcome never lands on the next lifecycle.
var generation = Volatile.Read(ref _lifecycleGeneration);
var session = Session;
if (session is null || (expectedSession is not null && !ReferenceEquals(session, expectedSession)))
{
@@ -1267,7 +1393,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp);
}
_health = new DriverHealth(DriverState.Healthy, now, null);
RecordDeviceAnswered(generation, now);
}
catch (Exception ex)
{
@@ -1280,7 +1406,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
var r = indexMap[w];
results[r] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
}
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
RecordCommsFault(generation, ex.Message);
return (results, false);
}
}
@@ -2062,6 +2188,29 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
UpdateHealth(h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null);
}
/// <summary>
/// A service call got a RESPONSE from the upstream server (whatever per-node statuses it carried — a refusal
/// is still the server answering): health Healthy, <see cref="DriverHealth.LastSuccessfulRead"/> advanced. Not
/// while a tracked subscription is known dead — that latch is what lets the comms-loss watchdog re-initialise
/// a driver whose session answers but whose data plane is gone.
/// </summary>
/// <param name="generation">The lifecycle generation the call ran under.</param>
/// <param name="now">When the response arrived.</param>
private void RecordDeviceAnswered(long generation, DateTime now)
{
if (Volatile.Read(ref _subscriptionLost) == 1) return;
UpdateHealth(generation, _ => new DriverHealth(DriverState.Healthy, now, null));
}
/// <summary>
/// A service call got NO response (transport failure / timeout): health Degraded with
/// <see cref="DriverHealth.LastSuccessfulRead"/> preserved, so the comms-loss watchdog measures the silence.
/// </summary>
/// <param name="generation">The lifecycle generation the call ran under.</param>
/// <param name="error">The failure, for <see cref="DriverHealth.LastError"/>.</param>
private void RecordCommsFault(long generation, string error) =>
UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, error));
/// <summary>
/// Read-modify-write of <see cref="_health"/> as a CAS loop: the keep-alive timer, publish
/// threads and the recovery task all derive the next health from the current one.
@@ -2078,6 +2227,27 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
}
/// <summary>
/// <see cref="UpdateHealth(Func{DriverHealth, DriverHealth?})"/> stamped with a lifecycle generation: the
/// write lands only while <paramref name="generation"/> is still current, re-checked on every CAS attempt. Every
/// lifecycle transition bumps the generation BEFORE it writes its own health, so a superseded writer (a late
/// keep-alive of a released session, a read that outlived a shutdown, an abandoned init) either loses the
/// generation check or is overwritten by the transition's own later write.
/// </summary>
/// <param name="generation">The lifecycle generation the writer belongs to.</param>
/// <param name="next">Maps the current health to the next one, or null to leave it unchanged.</param>
private void UpdateHealth(long generation, Func<DriverHealth, DriverHealth?> next) =>
UpdateHealth(h => Volatile.Read(ref _lifecycleGeneration) == generation ? next(h) : null);
/// <summary>
/// Bumped by every <see cref="InitializeAsync"/> and <see cref="ShutdownAsync"/>. Session install, the
/// namespace-map / endpoint publish and every health write carry the generation they were started under.
/// </summary>
private long _lifecycleGeneration;
/// <summary>Test seam: the current lifecycle generation.</summary>
internal long LifecycleGenerationForTest => Volatile.Read(ref _lifecycleGeneration);
/// <summary>
/// Live data-change subscription bookkeeping. Holds the SDK <see cref="Subscription"/>,
/// the local handle, and the per-MonitoredItem (item, handler) pairs so
@@ -2906,6 +3076,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// </remarks>
private void OnKeepAlive(ISession sender, KeepAliveEventArgs e)
{
// Captured BEFORE the session check: a shutdown / re-initialise that releases this session after the check
// has bumped the generation, so every health write below lands only for the session it was raised for.
var generation = Volatile.Read(ref _lifecycleGeneration);
// A late callback from a session already released (re-initialise) says nothing about the current one.
if (!ReferenceEquals(sender, Session)) return;
var now = Clock.GetUtcNow().UtcDateTime;
@@ -2923,7 +3096,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (Interlocked.Exchange(ref _sessionLost, 0) == 1)
{
UpdateHealth(_ => new DriverHealth(DriverState.Healthy, now, null));
UpdateHealth(generation, _ => 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);
var guardUntil = ArmReplayGuard(now);
@@ -2941,19 +3114,19 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (FindStalledSubscription(now) is { } stalled)
{
DeclareSubscriptionLost($"subscription '{stalled}' stopped publishing while the session stayed up", now,
DeclareSubscriptionLost(generation, $"subscription '{stalled}' stopped publishing while the session stayed up", now,
serverReported: false);
return;
}
UpdateHealth(h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null);
UpdateHealth(generation, h => h.State is DriverState.Healthy or DriverState.Degraded ? h with { LastSuccessfulRead = now } : null);
return;
}
Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0);
TransitionTo(HostState.Stopped);
_diag.Record(OpcUaClientDiagnostics.KeepAliveMiss);
UpdateHealth(h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"keep-alive: {e.Status}"));
UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"keep-alive: {e.Status}"));
// 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;
@@ -3058,15 +3231,16 @@ 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>
/// <param name="generation">The lifecycle generation the detection belongs to.</param>
/// <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 comms-loss watchdog.</param>
private void DeclareSubscriptionLost(string reason, DateTime now, bool serverReported)
private void DeclareSubscriptionLost(long generation, 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}"));
UpdateHealth(generation, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, $"subscription lost: {reason}"));
_logger.LogWarning(
"OpcUaClientDriver '{DriverInstanceId}': {Reason}; subscribed values are BadNoCommunication until the " +
"host re-initialises the driver and re-subscribes.", _driverInstanceId, reason);
@@ -3087,6 +3261,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
private void HandlePublishError(object? sender, StatusCode code, uint subscriptionId)
{
_diag.Record(OpcUaClientDiagnostics.PublishError);
var generation = Volatile.Read(ref _lifecycleGeneration);
if (sender is not null && !ReferenceEquals(sender, Session)) return;
bool tracked;
if (code == StatusCodes.BadNoSubscription)
@@ -3097,7 +3272,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
else
return;
if (tracked)
DeclareSubscriptionLost($"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime,
DeclareSubscriptionLost(generation, $"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime,
serverReported: true);
}
@@ -3349,6 +3524,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{
if (_disposed) return;
_disposed = true;
// Supersede any init still running, so it discards rather than installs its session.
Interlocked.Increment(ref _lifecycleGeneration);
lock (_probeLock)
{