From bfd5b7b015b73ae6d0c8854c18fab309b28e2775 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:26:45 -0400 Subject: [PATCH] fix(runtime): single-flight reconnect attempts, bounded pre-reconnect teardown (A2) The Reconnecting retry timer started a new ShutdownAsync + InitializeAsync pair on every tick whether or not the previous attempt had finished; the generation token only discarded the older RESULT, so with an Initialize bound longer than the cadence the teardown ran underneath an attempt still inside the driver (the dispose-before-overwrite race). An _initInFlight gate, cleared by the piped InitializeSucceeded/Failed, skips a tick while an attempt is running; the pre-reconnect teardown is bounded by the shutdown timeout; PostStop cancels an in-flight attempt before shutting the driver down. Review ref: 02/H2. Refs #731 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../Drivers/DriverInstanceActor.cs | 74 +++++++++- ...InstanceActorReconnectSingleFlightTests.cs | 126 ++++++++++++++++++ 2 files changed, 195 insertions(+), 5 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs 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); } + } +}