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

fix(opcuaclient): stale BadNoSubscription after a resubscribe is not a subscription loss (lmxopcua#763, lmxopcua#728)
v2-ci / unit-tests (push) Has been cancelled
Details

Browse Source 
HandlePublishError treated any anonymous BadNoSubscription (subscription id 0)
as loss of a tracked subscription whenever anything was tracked. A publish the
SDK queued while the session briefly held no subscription (the driver had just
deleted its last one) is answered BadNoSubscription and can be handled after the
replacement is registered, so a routine unsubscribe->resubscribe declared the new
subscription lost: Degraded health, a BadNoCommunication quality-only annotation
on every alarm condition and ConnectionLost (host re-initialise). That annotation
is the extra Unspecified event the #728 ConditionRefresh wire test saw.

- A BadNoSubscription that names a subscription id now counts only if the id is
  tracked (same rule as BadSubscriptionIdInvalid).
- An anonymous BadNoSubscription within StaleNoSubscriptionGrace (5 s) of a
  delete that left the driver tracking nothing is ignored (Debug log). The window
  is stamped on removal and restarted when the delete completes; failed-subscribe
  removes stamp it too; it resets with the session. A delete that leaves another
  subscription tracked opens no window. A genuine loss inside the window is still
  caught by the PublishingStopped heuristic; only the immediate ConnectionLost is
  forgone.
- OpcUaClientStaleNoSubscriptionTests: stale error after alarm and data
  resubscribe (fail without the fix), id-attributed stale error, plus genuine-loss
  controls (after the grace, no recent delete, non-last delete, session release),
  each mutation-checked.
- The ConditionRefresh wire test now records Kind, AlarmType, message, severity,
  quality, restatement and timestamp of every post-resubscribe event on failure.

Wire test locally: 1/10 passing before (all 9 failures the 0x80310000
ConditionType quality-only annotation), 12/12 after.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-11-09 12:47:58 -05:00
parent 73362a056a
commit 0217d36cf4
4 changed files with 297 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
+91 -9
View File
@@ -236,7 +236,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>Test-only seam: register an alarm subscription over <paramref name="refs"/> as
/// <see cref="SubscribeAlarmsAsync"/> would, without the wire create.</summary>
/// <param name="refs">The alarm RawPaths subscribed.</param>
internal void AddAlarmSubscriptionForTest(IReadOnlyList<string> refs)
/// <returns>The alarm subscription handle.</returns>
internal IAlarmSubscriptionHandle AddAlarmSubscriptionForTest(IReadOnlyList<string> refs)
{
var id = Interlocked.Increment(ref _nextAlarmSubscriptionId);
var handle = new OpcUaAlarmSubscriptionHandle(id);
@@ -245,6 +246,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_alarmSubscriptions[id] = new RemoteAlarmSubscription(
new Subscription(telemetry: null!, new SubscriptionOptions { DisplayName = $"{AlarmSubscriptionNamePrefix}{id}" }),
handle, item, (_, _) => { });
return handle;
}
/// <summary>
@@ -302,6 +304,26 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// (0 = none stalled). Only the keep-alive thread and the session release touch it.</summary>
private long _publishingStoppedSinceTicks;
/// <summary>
/// How long after the driver deleted its LAST server-side subscription an anonymous <c>BadNoSubscription</c>
/// (subscription id 0) is attributed to that delete rather than to a loss of whatever was subscribed since
/// (Gitea #763). A publish request the SDK had queued while the session held no subscription is answered
/// <c>BadNoSubscription</c>, and its PublishError can be handled AFTER the next subscription is registered
/// — an unsubscribe→resubscribe of the same set (DriverInstanceActor's reconcile) then looked like the
/// server dropping the new subscription. The SDK names no subscription on that status, so the delete is the
/// only thing it can be attributed to. Bounded on purpose: a genuine loss inside the window still degrades the
/// driver through the publishing-stall heuristic (<see cref="FindStalledSubscription"/>); only the immediate
/// <see cref="ConnectionLost"/> is forgone for it, and only within this window.
/// </summary>
internal static readonly TimeSpan StaleNoSubscriptionGrace = TimeSpan.FromSeconds(5);
/// <summary>
/// UTC ticks of the last time the driver deleted a subscription that left it tracking none (0 = never on this
/// session) — the moment the server could legitimately answer <c>BadNoSubscription</c>. See
/// <see cref="StaleNoSubscriptionGrace"/>. Reset when the session is released.
/// </summary>
private long _lastSubscriptionsEmptiedTicks;
/// <summary>The last in-place recovery started. Test seam: lets a test await it deterministically.</summary>
private Task _recoveryTask = Task.CompletedTask;
@@ -1191,6 +1213,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
Interlocked.Exchange(ref _sessionLost, 0);
Interlocked.Exchange(ref _subscriptionLost, 0);
Interlocked.Exchange(ref _publishingStoppedSinceTicks, 0);
Interlocked.Exchange(ref _lastSubscriptionsEmptiedTicks, 0);
// re-arm the definitive-loss signal for the next session.
Interlocked.Exchange(ref _connectionLostRaised, 0);
Interlocked.Exchange(ref _closedChannelFailures, 0);
@@ -1954,6 +1977,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
await session.RemoveSubscriptionAsync(subscription, removeCts.Token).ConfigureAwait(false);
}
catch { /* best-effort — the session may be the thing that failed */ }
// a half-built subscription can have been the server's only one, so its delete opens the same window.
NoteSubscriptionDeleted();
throw;
}
_subscriptions[id] = new RemoteSubscription(subscription, handle, itemHandlers);
@@ -2014,6 +2039,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{
if (handle is not OpcUaSubscriptionHandle h) return;
if (!_subscriptions.TryRemove(h.Id, out var rs)) return;
var emptied = NoteSubscriptionDeleted();
await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
@@ -2027,7 +2053,12 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort — the subscription may already be gone on reconnect */ }
}
finally { _subscriptionGate.Release(); }
finally
{
// the server answers the stale publish only once the delete is done, so the window runs from here.
if (emptied) NoteSubscriptionDeleted(force: true);
_subscriptionGate.Release();
}
}
private static void DetachNotificationHandlers(IReadOnlyList<MonitoredItemNotificationHandle> items)
@@ -2463,6 +2494,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
await session.RemoveSubscriptionAsync(subscription, removeCts.Token).ConfigureAwait(false);
}
catch { /* best-effort — the session may be the thing that failed */ }
// a half-built subscription can have been the server's only one, so its delete opens the same window.
NoteSubscriptionDeleted();
throw;
}
@@ -2512,6 +2545,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{
if (handle is not OpcUaAlarmSubscriptionHandle h) return;
if (!_alarmSubscriptions.TryRemove(h.Id, out var rs)) return;
var emptied = NoteSubscriptionDeleted();
await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
@@ -2524,7 +2558,12 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort — session may already be gone across a reconnect */ }
}
finally { _subscriptionGate.Release(); }
finally
{
// the server answers the stale publish only once the delete is done, so the window runs from here.
if (emptied) NoteSubscriptionDeleted(force: true);
_subscriptionGate.Release();
}
}
/// <inheritdoc />
@@ -3339,7 +3378,9 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>
/// Session PublishError: always counted; a BadNoSubscription / BadSubscriptionIdInvalid naming a
/// subscription the driver tracks means the server dropped it while the session lived.
/// subscription the driver tracks means the server dropped it while the session lived. An anonymous
/// BadNoSubscription means the server holds none at all — a loss of anything tracked, except within
/// <see cref="StaleNoSubscriptionGrace"/> of the driver deleting its last subscription (#763).
/// </summary>
private void OnPublishError(object? sender, PublishErrorEventArgs e) =>
HandlePublishError(sender, e.Status?.StatusCode ?? StatusCodes.Good, e.SubscriptionId);
@@ -3349,19 +3390,60 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_diag.Record(OpcUaClientDiagnostics.PublishError);
var generation = Volatile.Read(ref _lifecycleGeneration);
if (sender is not null && !ReferenceEquals(sender, Session)) return;
var now = Clock.GetUtcNow().UtcDateTime;
bool tracked;
if (code == StatusCodes.BadNoSubscription)
if (code == StatusCodes.BadSubscriptionIdInvalid
|| (code == StatusCodes.BadNoSubscription && subscriptionId != 0))
{
// The error names a subscription: it is a loss only if that subscription is one the driver still tracks
// (one it deleted itself, or never registered, is not).
tracked = IsTrackedSubscription(subscriptionId);
}
else if (code == StatusCodes.BadNoSubscription)
{
// Anonymous: the server holds NO subscription for the session. True of anything tracked — unless the
// driver just deleted its last subscription, when it is the answer to a publish queued in that gap (#763).
tracked = !_subscriptions.IsEmpty || !_alarmSubscriptions.IsEmpty;
else if (code == StatusCodes.BadSubscriptionIdInvalid)
tracked = _subscriptions.Values.Any(r => r.Subscription.Id == subscriptionId)
|| _alarmSubscriptions.Values.Any(r => r.Subscription.Id == subscriptionId);
if (tracked && IsWithinStaleNoSubscriptionGrace(now))
{
_logger.LogDebug(
"OpcUaClientDriver '{DriverInstanceId}': ignoring BadNoSubscription within {Grace} of deleting the " +
"last subscription — it answers a publish queued while the session held none, not a loss of the " +
"subscription registered since.", _driverInstanceId, StaleNoSubscriptionGrace);
return;
}
}
else
return;
if (tracked)
DeclareSubscriptionLost(generation, $"server reported {code} for a tracked subscription", Clock.GetUtcNow().UtcDateTime,
DeclareSubscriptionLost(generation, $"server reported {code} for a tracked subscription", now,
serverReported: true);
}
private bool IsTrackedSubscription(uint subscriptionId) =>
_subscriptions.Values.Any(r => r.Subscription.Id == subscriptionId)
|| _alarmSubscriptions.Values.Any(r => r.Subscription.Id == subscriptionId);
/// <summary>
/// Stamp <see cref="_lastSubscriptionsEmptiedTicks"/> when a delete leaves the driver tracking no subscription
/// (or unconditionally with <paramref name="force"/>, to restart the window once that delete has completed).
/// </summary>
/// <param name="force">Stamp even though a subscription has been registered since (the delete it follows did
/// empty the set).</param>
/// <returns>True when the stamp was taken.</returns>
private bool NoteSubscriptionDeleted(bool force = false)
{
if (!force && (!_subscriptions.IsEmpty || !_alarmSubscriptions.IsEmpty)) return false;
Interlocked.Exchange(ref _lastSubscriptionsEmptiedTicks, Clock.GetUtcNow().UtcDateTime.Ticks);
return true;
}
private bool IsWithinStaleNoSubscriptionGrace(DateTime now)
{
var emptied = Interlocked.Read(ref _lastSubscriptionsEmptiedTicks);
return emptied != 0 && now.Ticks - emptied < StaleNoSubscriptionGrace.Ticks;
}
/// <summary>
/// publish the start of an outage: <c>BadNoCommunication</c> once per (subscription, ref) with the last
/// delivered value retained, and a quality-only annotation on every subscribed alarm condition (no state