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

fix(runtime): single-flight reconnect attempts, bounded pre-reconnect teardown (A2)

Browse Source 
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
This commit is contained in:
Joseph Doherty
2026-10-22 06:26:45 -04:00
parent eb23053700
commit bfd5b7b015
2 changed files with 195 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
+69 -5
View File
@@ -240,6 +240,21 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// on the actor thread, so no lock is needed.</summary>
private int _initGeneration;
/// <summary>
/// True while an <see cref="InitializeAsync"/> attempt is running (set when it is launched, cleared by its
/// piped <see cref="InitializeSucceeded"/>/<see cref="InitializeFailed"/>). The <c>Reconnecting</c> 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.
/// </summary>
private bool _initInFlight;
/// <summary>Cancellation for the in-flight attempt, so <see cref="PostStop"/> can cancel it before the
/// shutdown rather than let it race the teardown on a restart.</summary>
private CancellationTokenSource? _initCts;
/// <summary>
/// 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<InitializeSucceeded>(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<InitializeFailed>(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<InitializeSucceeded>(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<InitializeFailed>(_ => { /* 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<InitializeFailed>(msg =>
{
if (msg.Generation == _initGeneration) CompleteInitAttempt();
});
Receive<SetDesiredSubscriptions>(StoreDesiredSubscriptions);
Receive<ForceReconnect>(_ => { /* already reconnecting — no-op */ });
Receive<ConnectionLossRaised>(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();
}
});
}
/// <summary>Releases the single-flight gate once the in-flight attempt's result has been honoured.</summary>
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.
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs
+126
View File
@@ -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;
/// <summary>
/// Review 02 H2: the <c>Reconnecting</c> retry timer fired every <c>reconnectInterval</c> whether or not the
/// previous attempt was still running — <c>_initGeneration</c> 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 <c>ShutdownAsync</c> ran while attempt N's <c>InitializeAsync</c> 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.
/// </summary>
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);
}
/// <summary>Records every Shutdown/Initialize entry and exit so an overlap is a fact, not an inference.</summary>
private sealed class InterleavingDriver : IDriver
{
private readonly TimeSpan _initDuration;
private readonly int _failUntilAttempt;
private readonly Lock _gate = new();
private readonly List<string> _events = new();
private int _attempts;
private int _initsInProgress;
private int _overlaps;
public InterleavingDriver(TimeSpan initDuration, int failUntilAttempt)
{
_initDuration = initDuration;
_failUntilAttempt = failUntilAttempt;
}
public Func<CancellationToken, Task>? 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($"<init{attempt}");
Interlocked.Decrement(ref _initsInProgress);
}
}
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => 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("<shutdown");
}
}
public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);
private void Record(string e) { lock (_gate) _events.Add(e); }
}
}