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

fix(drivers): InitializeAsync releases previous device/session state before overwrite (#630)

Browse Source 
FOCAS, AbLegacy, AbCip, TwinCAT, S7 and OpcUaClient orphaned the previous
probe/recycle/fixed-tree loops and clients when re-initialised on a live driver.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 14:28:55 -04:00
parent d898262c4a
commit 461133235a
13 changed files with 494 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs
+35 -2
View File
@@ -335,8 +335,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
public string DriverType => "AbCip";
/// <inheritdoc />
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// #630: InitializeAsync is reachable on a live driver (DriverInstanceActor.AdoptConfigDuringInit
// runs it with no teardown first). Release the previous device states' probe loops + handles before
// the overwrite below. Deliberately NOT ShutdownAsync: that also drops _poll and the subscriptions.
await ReleaseDeviceStatesAsync(cancellationToken).ConfigureAwait(false);
_health = new DriverHealth(DriverState.Initializing, null, null);
_logger.LogInformation(
"AbCip driver {DriverInstanceId} initializing: transport={Transport} tagFactory={TagFactory}",
@@ -459,7 +463,36 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
_logger.LogError(ex, "AbCip driver {DriverInstanceId} failed to initialize", _driverInstanceId);
throw;
}
return Task.CompletedTask;
}
/// <summary>
/// Cancels, joins (bounded) and disposes every existing device state's probe loop + handles, then
/// clears the device/route tables. Device part of <see cref="ShutdownAsync"/> only.
/// </summary>
private async Task ReleaseDeviceStatesAsync(CancellationToken cancellationToken)
{
if (_devices.Count == 0) return;
foreach (var state in _devices.Values)
{
try { state.ProbeCts?.Cancel(); } catch (ObjectDisposedException) { }
}
foreach (var state in _devices.Values)
{
var probeTask = state.ProbeTask;
if (probeTask is null) continue;
try { await probeTask.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false); }
catch (TimeoutException) { }
catch (OperationCanceledException) { }
}
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeHandles();
}
_devices.Clear();
_devicesByRoutingKey.Clear();
}
/// <inheritdoc />
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
+35 -2
View File
@@ -143,8 +143,13 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
}
/// <inheritdoc />
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// #630: InitializeAsync can run on a live driver (DriverInstanceActor.AdoptConfigDuringInit calls it
// without a teardown first). Release the previous device states' probe loops + runtimes BEFORE the
// overwrite below, or they are orphaned beyond the reach of ShutdownAsync. Deliberately NOT
// ShutdownAsync: that also drops _poll and every subscription, which nothing would re-drive.
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
_health = new DriverHealth(DriverState.Initializing, null, null);
_logger.LogInformation(
"AbLegacy driver {DriverInstanceId} initializing: transport={Transport} tagFactory={TagFactory}",
@@ -255,7 +260,35 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
_resolver.Clear(); // drop transient equipment-tag parses so a config change re-parses
throw;
}
return Task.CompletedTask;
}
/// <summary>
/// Cancels, bounded-joins and disposes every existing device state, then clears the table.
/// Device-state part of <see cref="ShutdownAsync"/> only (no poll / subscription teardown).
/// </summary>
private async Task ReleaseDeviceStatesAsync()
{
if (_devices.Count == 0) return;
foreach (var state in _devices.Values)
try { state.ProbeCts?.Cancel(); } catch { }
var loops = _devices.Values.Select(d => d.ProbeTask).OfType<Task>().ToList();
if (loops.Count > 0)
{
var all = Task.WhenAll(loops);
var finished = await Task.WhenAny(all, Task.Delay(LoopJoinTimeout, CancellationToken.None)).ConfigureAwait(false);
if (finished != all)
_logger.LogWarning(
"AbLegacy re-initialise: {Count} probe loop(s) did not stop within {Timeout}; proceeding. Driver={DriverInstanceId}",
loops.Count, LoopJoinTimeout, _driverInstanceId);
}
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeRuntimes();
}
_devices.Clear();
}
/// <inheritdoc />
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs
+55 -2
View File
@@ -178,8 +178,12 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
/// <param name="driverConfigJson">The driver configuration JSON (see remarks — not re-parsed).</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A completed task.</returns>
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// #630: release the previous device states' loops + clients BEFORE they are overwritten (the actor's
// AdoptConfigDuringInit calls this without a teardown first). Not ShutdownAsync: that also drops _poll
// and subscriptions, which nothing would re-drive.
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
Volatile.Write(ref _health, new DriverHealth(DriverState.Initializing, null, null));
try
{
@@ -316,7 +320,56 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
Volatile.Write(ref _health, new DriverHealth(DriverState.Faulted, null, ex.Message));
throw;
}
return Task.CompletedTask;
}
/// <summary>
/// Cancels, bounded-joins and disposes every existing device state (probe / recycle / fixed-tree
/// loops + client) and the alarm projection, then clears the table. Device part of
/// <see cref="ShutdownAsync"/> only (no poll / subscription teardown).
/// </summary>
private async Task ReleaseDeviceStatesAsync()
{
if (_alarmProjection is { } proj)
{
await proj.DisposeAsync().ConfigureAwait(false);
_alarmProjection = null;
}
if (_devices.Count == 0) return;
var loops = new List<Task>();
foreach (var state in _devices.Values)
{
try { state.ProbeCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling probe CTS for {Host} failed", state.Options.HostAddress); }
try { state.RecycleCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling recycle CTS for {Host} failed", state.Options.HostAddress); }
try { state.FixedTreeCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling fixed-tree CTS for {Host} failed", state.Options.HostAddress); }
if (state.ProbeTask is { } pt) loops.Add(pt);
if (state.RecycleTask is { } rt) loops.Add(rt);
if (state.FixedTreeTask is { } ft) loops.Add(ft);
}
if (loops.Count > 0)
{
var all = Task.WhenAll(loops);
var finished = await Task.WhenAny(all, Task.Delay(LoopJoinTimeout, CancellationToken.None)).ConfigureAwait(false);
if (finished != all)
_logger.LogWarning("FOCAS re-initialise: {Count} background loop(s) did not stop within {Timeout}; proceeding",
loops.Count, LoopJoinTimeout);
}
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.RecycleCts?.Dispose();
state.RecycleCts = null;
state.RecycleTask = null;
state.FixedTreeCts?.Dispose();
state.FixedTreeCts = null;
state.FixedTreeTask = null;
state.DisposeClient();
}
_devices.Clear();
}
/// <inheritdoc />
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
+40 -25
View File
@@ -194,6 +194,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <inheritdoc />
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// #630: InitializeAsync is reachable on a live driver (DriverInstanceActor.AdoptConfigDuringInit runs
// it with no teardown first); `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);
try
{
@@ -587,32 +591,14 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
/// <inheritdoc />
public async Task ShutdownAsync(CancellationToken cancellationToken)
/// <summary>
/// Cancels any in-flight reconnect, 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 (#630) does not orphan the
/// previous session and its reconnect handler. Subscriptions are NOT touched here.
/// </summary>
private async Task ReleaseSessionAsync(CancellationToken cancellationToken)
{
// 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
// after close anyway. Before deleting each subscription we detach the Notification
// handlers we attached at subscribe time so the SDK's invocation list no longer
// holds the driver instance through the closure.
foreach (var rs in _subscriptions.Values)
{
DetachNotificationHandlers(rs.ItemHandlers);
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
}
_subscriptions.Clear();
foreach (var ras in _alarmSubscriptions.Values)
{
try { ras.EventItem.Notification -= ras.Handler; }
catch { /* best-effort */ }
try { await ras.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
}
_alarmSubscriptions.Clear();
// Abort any in-flight reconnect attempts before touching the session — BeginReconnect's
// retry loop holds a reference to the current session and would fight Session.CloseAsync
// if left spinning. Take the handler under _probeLock so a keep-alive callback racing
@@ -645,6 +631,35 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try { if (sessionToClose is Session s) await s.CloseAsync(cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
try { sessionToClose?.Dispose(); } catch { }
}
/// <inheritdoc />
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
// 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
// after close anyway. Before deleting each subscription we detach the Notification
// handlers we attached at subscribe time so the SDK's invocation list no longer
// holds the driver instance through the closure.
foreach (var rs in _subscriptions.Values)
{
DetachNotificationHandlers(rs.ItemHandlers);
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
}
_subscriptions.Clear();
foreach (var ras in _alarmSubscriptions.Values)
{
try { ras.EventItem.Notification -= ras.Handler; }
catch { /* best-effort */ }
try { await ras.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
}
_alarmSubscriptions.Clear();
await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false);
_namespaceMap = null;
_connectedEndpointUrl = null;
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs
+16 -3
View File
@@ -223,6 +223,11 @@ public sealed class S7Driver
/// <inheritdoc />
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// #630: InitializeAsync is reachable on a live driver (DriverInstanceActor.AdoptConfigDuringInit
// runs it with no teardown first). Stop the previous probe loop and drop the previous connection
// before they are overwritten below. Deliberately NOT ShutdownAsync: that also drops _poll and the
// subscriptions, which nothing would re-drive.
await ReleaseConnectionAsync().ConfigureAwait(false);
Health = new DriverHealth(DriverState.Initializing, null, null);
try
{
@@ -360,7 +365,18 @@ public sealed class S7Driver
// disposing its CTS (drain-before-dispose, STAB-6-S7), so no poll iteration touches _gate
// after this returns.
await _poll.DisposeAsync().ConfigureAwait(false);
await ReleaseConnectionAsync().ConfigureAwait(false);
_writeFault = null;
_connectBackoff.RecordSuccess();
Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null);
}
/// <summary>
/// Cancels + bounded-joins the probe loop, then disposes its CTS and the PLC connection. The
/// connection part of <see cref="ShutdownAsync"/> (everything but the poll engine and health).
/// </summary>
private async Task ReleaseConnectionAsync()
{
var probeCts = _probeCts;
var probeTask = _probeTask;
try { probeCts?.Cancel(); } catch { }
@@ -381,11 +397,8 @@ public sealed class S7Driver
_initialized = false;
_plcDead = false;
_writeFault = null;
_connectBackoff.RecordSuccess();
try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ }
Plc = null;
Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null);
}
/// <inheritdoc />
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs
+29 -2
View File
@@ -110,8 +110,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
public string DriverType => "TwinCAT";
/// <inheritdoc />
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// #630: InitializeAsync is reachable on a live driver (DriverInstanceActor.AdoptConfigDuringInit
// runs it with no teardown first). Release the previous device states' probe loops + clients before
// the overwrite below. Deliberately NOT ShutdownAsync: that also drops _poll and the subscriptions.
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
_health = new DriverHealth(DriverState.Initializing, null, null);
try
{
@@ -175,7 +179,30 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
throw;
}
return Task.CompletedTask;
}
/// <summary>
/// Cancels, awaits and disposes every existing device state's probe loop, client and gate, then
/// clears the device table. Device part of <see cref="ShutdownAsync"/> only.
/// </summary>
private async Task ReleaseDeviceStatesAsync()
{
if (_devices.IsEmpty) return;
foreach (var state in _devices.Values)
{
try { state.ProbeCts?.Cancel(); } catch { }
if (state.ProbeTask is Task pt)
{
try { await pt.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); }
catch { /* cancelled, faulted or wedged — teardown continues */ }
}
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeClient();
state.DisposeGate();
}
_devices.Clear();
}
/// <inheritdoc />