diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
index 5610cb1f..f666fd28 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
@@ -240,6 +240,21 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// on the actor thread, so no lock is needed.
private int _initGeneration;
+ ///
+ /// True while an attempt is running (set when it is launched, cleared by its
+ /// piped /). The Reconnecting retry
+ /// timer fires on its cadence regardless of whether the previous attempt finished, and the generation
+ /// token only discards an older attempt's RESULT — so with an Initialize bound longer than the cadence
+ /// (Tier C: 15 s vs. 10 s) attempt N+1's teardown ran while attempt N was still inside the driver: the
+ /// dispose-before-overwrite race the drivers' rule file forbids, once per overlap. A tick that lands while
+ /// an attempt is in flight is skipped; the next tick after the result retries. Actor-thread only.
+ ///
+ private bool _initInFlight;
+
+ /// Cancellation for the in-flight attempt, so can cancel it before the
+ /// shutdown rather than let it race the teardown on a restart.
+ private CancellationTokenSource? _initCts;
+
///
/// Timestamps of recent Faulted-state transitions; used to compute the 5-minute error count.
/// No lock needed — every read/write site runs inside an Akka message handler, which is
@@ -498,6 +513,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Receive(msg =>
{
if (msg.Generation != _initGeneration) return;
+ CompleteInitAttempt();
_log.Info("DriverInstance {Id}: connected", _driverInstanceId);
Become(Connected);
PublishHealthSnapshot();
@@ -508,6 +524,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Receive(msg =>
{
if (msg.Generation != _initGeneration) return;
+ CompleteInitAttempt();
_log.Warning("DriverInstance {Id}: initialize failed: {Reason}", _driverInstanceId, msg.Reason);
RecordFault();
Become(Reconnecting);
@@ -795,6 +812,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Receive(msg =>
{
if (msg.Generation != _initGeneration) return;
+ CompleteInitAttempt();
Timers.Cancel("retry-connect");
_log.Info("DriverInstance {Id}: reconnected", _driverInstanceId);
Become(Connected);
@@ -803,9 +821,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
AttachAlarmSource();
SubscribeDesiredAlarms();
});
- // A failure here is a no-op regardless of generation — the retry timer keeps trying the
- // current config; only a (generation-matched) InitializeSucceeded transitions state.
- Receive(_ => { /* keep retrying via timer */ });
+ // A failure here changes no state — the retry timer keeps trying the current config; only a
+ // (generation-matched) InitializeSucceeded transitions. It DOES release the single-flight gate, or
+ // no later tick could ever retry.
+ Receive(msg =>
+ {
+ if (msg.Generation == _initGeneration) CompleteInitAttempt();
+ });
Receive(StoreDesiredSubscriptions);
Receive(_ => { /* already reconnecting — no-op */ });
Receive(IgnoreConnectionLoss);
@@ -850,12 +872,26 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
private void InitializeAsync(string driverConfigJson, bool teardownFirst = false)
{
_currentConfigJson = driverConfigJson;
+ if (_initInFlight)
+ {
+ // SINGLE-FLIGHT (02/H2). The previous attempt is still inside the driver; starting another would run
+ // its teardown against the transport that attempt is building. Skip — the next tick after the
+ // result retries, and a config handed over meanwhile is already stored for it.
+ _log.Debug("DriverInstance {Id}: initialise attempt {Gen} still in flight; skipping this one",
+ _driverInstanceId, _initGeneration);
+ return;
+ }
+
+ _initInFlight = true;
var generation = ++_initGeneration;
+ var initCts = new CancellationTokenSource();
+ _initCts = initCts;
var self = Self;
var log = _log;
var driverInstanceId = _driverInstanceId;
var driver = _driver;
var invoker = _invoker;
+ var shutdownTimeout = _shutdownTimeout;
_ = Task.Run(async () =>
{
try
@@ -866,7 +902,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
{
// Deliberately UNWRAPPED: teardown is best-effort cleanup of an already-dead
// transport — a shutdown failure must not count toward the Initialize breaker.
- await driver.ShutdownAsync(CancellationToken.None);
+ // BOUNDED: a teardown wedged on the dead transport must not hold the retry loop
+ // (and this thread) for longer than the bound; it is abandoned and the init proceeds.
+ using var teardownCts = new CancellationTokenSource(shutdownTimeout);
+ await driver.ShutdownAsync(teardownCts.Token).WaitAsync(shutdownTimeout);
+ }
+ catch (TimeoutException)
+ {
+ log.Warning(
+ "DriverInstance {Id}: ShutdownAsync did not complete within {Bound} before a reconnect " +
+ "attempt; abandoning it and initialising anyway", driverInstanceId, shutdownTimeout);
}
catch (Exception ex)
{
@@ -884,20 +929,33 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// attempt with BrokenCircuitException, which lands in the catch below exactly like a
// real connect failure: InitializeFailed → keep retrying on the timer. The 15 s break
// window is shorter than two 10 s timer ticks, so recovery costs at most one extra tick.
+ // The outer token is this attempt's own: PostStop cancels it so a stop/restart does not run its
+ // shutdown underneath an attempt still inside the driver.
await invoker.ExecuteAsync(
DriverCapability.Initialize,
driverInstanceId,
async ct => await driver.InitializeAsync(driverConfigJson, ct),
- CancellationToken.None);
+ initCts.Token);
self.Tell(new InitializeSucceeded(generation));
}
catch (Exception ex)
{
self.Tell(new InitializeFailed(ex.Message, generation));
}
+ finally
+ {
+ initCts.Dispose();
+ }
});
}
+ /// Releases the single-flight gate once the in-flight attempt's result has been honoured.
+ private void CompleteInitAttempt()
+ {
+ _initInFlight = false;
+ _initCts = null;
+ }
+
private async Task HandleWriteAsync(WriteAttribute msg)
{
var replyTo = Sender;
@@ -1591,6 +1649,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
DetachHostStatusSource();
DetachConnectionLossSource();
+ // An attempt still inside the driver is cancelled before the shutdown below runs underneath it (on a
+ // restart the next incarnation re-initialises the same object). Best-effort: the task owns the CTS
+ // and may have disposed it already.
+ try { _initCts?.Cancel(); }
+ catch (ObjectDisposedException) { /* the attempt already finished */ }
+
// BOUNDED. PostStop runs on a dispatcher thread and this blocks it, so an unbounded wait on a driver
// whose shutdown hangs against a dead socket / COM proxy pinned that thread for the whole wait, and a
// few such stops starved the dispatcher. A shutdown that overruns the bound is abandoned.
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs
new file mode 100644
index 00000000..19f1b6e3
--- /dev/null
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs
@@ -0,0 +1,126 @@
+using Akka.Actor;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+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;
+
+///
+/// Review 02 H2: the Reconnecting retry timer fired every reconnectInterval whether or not the
+/// previous attempt was still running — _initGeneration only discarded the older attempt's RESULT. With an
+/// Initialize bound longer than the cadence (Tier C's 15 s vs. the 10 s timer, or the test harness's unbounded
+/// invoker) attempt N+1's ShutdownAsync ran while attempt N's InitializeAsync was still inside the
+/// driver: the dispose-before-overwrite race the drivers' rule file forbids, once per overlap, for as long as
+/// the device stayed down. The retry is now single-flight.
+///
+public sealed class DriverInstanceActorReconnectSingleFlightTests : RuntimeActorTestBase
+{
+ [Fact]
+ public void Shutdown_and_Initialize_never_overlap_while_the_init_outlasts_the_retry_cadence()
+ {
+ // A 400 ms failing init against a 50 ms retry cadence: without the gate the timer starts a new
+ // Shutdown+Initialize pair every 50 ms on top of the one still running (eight-deep overlap).
+ var driver = new InterleavingDriver(initDuration: TimeSpan.FromMilliseconds(400), failUntilAttempt: 4);
+ var parent = CreateTestProbe();
+ var actor = parent.ChildActorOf(DriverInstanceActor.Props(
+ driver, reconnectInterval: TimeSpan.FromMilliseconds(50), commsLossTimeout: TimeSpan.Zero));
+
+ actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
+
+ // Positive control: the gate skips ticks, it does not stop retrying — the 4th attempt succeeds and the
+ // actor connects.
+ parent.FishForMessage(m => m is DriverInstanceActor.ConnectivityChanged { Connected: true }, PresenceBudget);
+
+ driver.Attempts.ShouldBeGreaterThanOrEqualTo(4);
+ driver.Overlaps.ShouldBe(0,
+ $"ShutdownAsync entered while an InitializeAsync was still running. Interleaving: {driver.Interleaving}");
+ }
+
+ [Fact]
+ public void A_pre_reconnect_teardown_that_hangs_is_abandoned_at_the_bound_and_the_init_still_runs()
+ {
+ var never = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var driver = new InterleavingDriver(initDuration: TimeSpan.Zero, failUntilAttempt: 2)
+ {
+ // Only the SECOND shutdown (the reconnect teardown) hangs; the first attempt has none.
+ ShutdownBehavior = _ => never.Task,
+ };
+ var parent = CreateTestProbe();
+ var actor = parent.ChildActorOf(DriverInstanceActor.Props(
+ driver, reconnectInterval: TimeSpan.FromMilliseconds(50), commsLossTimeout: TimeSpan.Zero,
+ shutdownTimeout: TimeSpan.FromMilliseconds(200)));
+
+ actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
+
+ parent.FishForMessage(m => m is DriverInstanceActor.ConnectivityChanged { Connected: true }, PresenceBudget);
+ never.Task.IsCompleted.ShouldBeFalse("the teardown is still hung — the attempt did not wait for it");
+ driver.Attempts.ShouldBe(2);
+ }
+
+ /// Records every Shutdown/Initialize entry and exit so an overlap is a fact, not an inference.
+ private sealed class InterleavingDriver : IDriver
+ {
+ private readonly TimeSpan _initDuration;
+ private readonly int _failUntilAttempt;
+ private readonly Lock _gate = new();
+ private readonly List _events = new();
+ private int _attempts;
+ private int _initsInProgress;
+ private int _overlaps;
+
+ public InterleavingDriver(TimeSpan initDuration, int failUntilAttempt)
+ {
+ _initDuration = initDuration;
+ _failUntilAttempt = failUntilAttempt;
+ }
+
+ public Func? ShutdownBehavior { get; init; }
+
+ public int Attempts => Volatile.Read(ref _attempts);
+ public int Overlaps => Volatile.Read(ref _overlaps);
+ public string Interleaving { get { lock (_gate) return string.Join(" ", _events); } }
+
+ public string DriverInstanceId => "interleaving-1";
+ public string DriverType => "Interleaving";
+
+ public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
+ {
+ var attempt = Interlocked.Increment(ref _attempts);
+ Interlocked.Increment(ref _initsInProgress);
+ Record($"init{attempt}>");
+ try
+ {
+ if (_initDuration > TimeSpan.Zero) await Task.Delay(_initDuration, CancellationToken.None);
+ if (attempt < _failUntilAttempt) throw new InvalidOperationException($"init attempt {attempt} refused");
+ }
+ finally
+ {
+ Record($" Task.CompletedTask;
+
+ public async Task ShutdownAsync(CancellationToken cancellationToken)
+ {
+ if (Volatile.Read(ref _initsInProgress) > 0) Interlocked.Increment(ref _overlaps);
+ Record("shutdown>");
+ try
+ {
+ if (ShutdownBehavior is { } behavior) await behavior(cancellationToken);
+ }
+ finally
+ {
+ Record(" new(DriverState.Healthy, DateTime.UtcNow, null);
+
+ private void Record(string e) { lock (_gate) _events.Add(e); }
+ }
+}