fix(drivers): #630 review - AbCip disposes alarm projection + clears tag tables, join not abortable by init token; S7 swaps Plc under _gate; doc rationale

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 14:39:05 -04:00
parent 461133235a
commit 18a20981a5
6 changed files with 56 additions and 10 deletions
+1 -1
View File
@@ -432,7 +432,7 @@ the previous state's probe / recycle / fixed-tree loops and its client. FOCAS, A
OpcUaClient now cancel, bounded-join (5 s) and dispose that state at the top of `InitializeAsync`
(`ReleaseDeviceStatesAsync` / `ReleaseConnectionAsync` / `ReleaseSessionAsync`), alongside Modbus and MQTT.
It is deliberately the device-state half of `ShutdownAsync` only: the poll engine and subscriptions are *not*
dropped, because nothing would re-drive them. A new driver must do the same.
dropped: the release is scoped to device state so it cannot take state the shutdown path owns. The actor re-subscribes values and alarms after every successful init, so driver-side subscriptions kept across a no-teardown re-init are retained leftovers, not needed for data flow. A new driver must do the same.
#### The fan-out that watchdog triggers: indexed per driver, and one batch
@@ -471,7 +471,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
/// </summary>
private async Task ReleaseDeviceStatesAsync(CancellationToken cancellationToken)
{
if (_devices.Count == 0) return;
// The projection is replaced (not reused) when a config body is adopted; dispose the old one so its
// poll loop does not outlive it and raise duplicate alarm events through this driver.
await _alarmProjection.DisposeAsync().ConfigureAwait(false);
foreach (var state in _devices.Values)
{
try { state.ProbeCts?.Cancel(); } catch (ObjectDisposedException) { }
@@ -480,7 +482,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
{
var probeTask = state.ProbeTask;
if (probeTask is null) continue;
try { await probeTask.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false); }
// Own bounded timeout, independent of the caller token: cancelling init must not abort the join
// and dispose handles under a probe loop that is still running.
try { await probeTask.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); }
catch (TimeoutException) { }
catch (OperationCanceledException) { }
}
@@ -493,6 +497,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
}
_devices.Clear();
_devicesByRoutingKey.Clear();
// Rebuilt wholesale by Initialize; stale entries would trip its member fan-out collision check.
_tagsByRawPath.Clear();
_declaredTags.Clear();
}
/// <inheritdoc />
@@ -148,7 +148,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
// #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.
// ShutdownAsync: that also drops _poll and every subscription, which are owned by the shutdown path (the actor re-subscribes after every successful init).
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
_health = new DriverHealth(DriverState.Initializing, null, null);
_logger.LogInformation(
@@ -182,7 +182,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
{
// #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.
// and subscriptions, which are owned by the shutdown path (the actor re-subscribes after every successful init).
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
Volatile.Write(ref _health, new DriverHealth(DriverState.Initializing, null, null));
try
@@ -226,7 +226,7 @@ public sealed class S7Driver
// #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.
// subscriptions, which are owned by the shutdown path (the actor re-subscribes after every successful init).
await ReleaseConnectionAsync().ConfigureAwait(false);
Health = new DriverHealth(DriverState.Initializing, null, null);
try
@@ -395,10 +395,22 @@ public sealed class S7Driver
_probeCts = null;
_probeTask = null;
_initialized = false;
_plcDead = false;
try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ }
Plc = null;
// Swap the connection out under _gate so a Read/Write/poll pass that already holds the gate (they
// re-read Plc inside it) can never see a disposed Plc mid-swap. Bounded: a wedged holder must not hang
// teardown, in which case we proceed exactly as the un-gated path always did.
var gated = false;
try { gated = await _gate.WaitAsync(DrainTimeout).ConfigureAwait(false); } catch { }
try
{
_initialized = false;
_plcDead = false;
try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ }
Plc = null;
}
finally
{
if (gated) { try { _gate.Release(); } catch (ObjectDisposedException) { } }
}
}
/// <inheritdoc />
@@ -229,6 +229,33 @@ public sealed class AbCipAlarmProjectionTests
}
/// <summary>Waits for <paramref name="more"/> further reads of the fake runtime (a processed-count wait).</summary>
/// <summary>
/// #630 — a re-Initialize on a live driver must dispose the alarm projection's poll loop rather
/// than leave it running (it would keep raising duplicate alarm events through this driver).
/// </summary>
[Fact]
public async Task Reinitialize_Stops_The_Previous_Projection_Poll_Loop()
{
var factory = new FakeAbCipTagFactory();
var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None);
await WaitForTagCreation(factory, "HighTemp");
var tag = factory.Tags["HighTemp"];
// Positive control: the loop is demonstrably polling before the re-init.
await WaitForMoreReadsAsync(factory, "HighTemp", 3);
await drv.InitializeAsync("{}", CancellationToken.None);
int Reads() => factory.Tags.Values.Sum(t => t.ReadCount); // the loop may re-create its runtime on the new device state
var frozen = Reads();
await Task.Delay(300); // ~15 poll intervals; a live loop would advance the count
Reads().ShouldBe(frozen);
await drv.ShutdownAsync(CancellationToken.None);
}
private static async Task WaitForMoreReadsAsync(FakeAbCipTagFactory factory, string tagName, int more)
{
var tag = factory.Tags[tagName];