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

fix(runtime): pessimistic write bound, incarnation-stamped write completions (fix round 2)

Browse Source 
- RunWriteAsync wraps the invoker call in WaitAsync(writeTimeout) (Props,
  default 10 s = 2x the 5 s write cap): a token-ignoring hung write now
  answers BadTimeout, releases its slot and tag, and its orphan outcome
  is observed and discarded - later writes to that tag reach the device.
- WriteCompleted carries the launching incarnation (from the process-wide
  generation source); a late completion from a previous incarnation after
  a supervision restart is ignored instead of freeing the new
  incarnation's slot/tag.
- Test: a hung token-ignoring tag-a write is abandoned and a later tag-a
  write reaches the device while the abandoned one is still inside it.

Refs #731

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 10:43:45 -04:00
parent 953bd27533
commit 7ae74ef147
2 changed files with 103 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
+65 -12
View File
@@ -218,8 +218,18 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// </summary>
public static readonly TimeSpan DefaultInitializeTimeout = TimeSpan.FromSeconds(30);
/// <summary>
/// Pessimistic bound on one write round-trip, on top of the 5 s token-based write cap. A driver whose
/// <see cref="IWritable.WriteAsync"/> ignores its token would otherwise hold its slot AND its tag forever,
/// piling every later write to that tag up in the queue. 2× the write cap.
/// </summary>
public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(10);
/// <summary>Self-sent by a write task once it has replied, releasing its slot and its tag.</summary>
private sealed record WriteCompleted(string TagId);
/// <param name="TagId">The tag the write held.</param>
/// <param name="Incarnation">The actor incarnation that launched it — a late completion from a previous
/// incarnation (after a supervision restart) must not free the new incarnation's slot or tag.</param>
private sealed record WriteCompleted(string TagId, int Incarnation);
private readonly IDriver _driver;
@@ -266,6 +276,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <summary>See <see cref="DefaultInitializeTimeout"/>.</summary>
private readonly TimeSpan _initializeTimeout;
/// <summary>See <see cref="DefaultWriteTimeout"/>.</summary>
private readonly TimeSpan _writeTimeout;
/// <summary>This actor object's identity for its own off-mailbox completions (drawn from the process-wide
/// generation source, so it is unique across supervision restarts of the same child).</summary>
private readonly int _incarnation;
/// <summary>
/// Set by <see cref="PreRestart"/> so the <see cref="PostStop"/> it triggers skips the dispose. A
/// supervision restart recreates the actor object around the SAME <see cref="IDriver"/> (the Props closure
@@ -407,6 +424,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <c>BadTimeout</c>; defaults to <see cref="DefaultQueuedWriteBudget"/>.</param>
/// <param name="initializeTimeout">Optional pessimistic bound on one initialise attempt; defaults to
/// <see cref="DefaultInitializeTimeout"/>.</param>
/// <param name="writeTimeout">Optional pessimistic bound on one write round-trip; defaults to
/// <see cref="DefaultWriteTimeout"/>.</param>
/// <returns>A <see cref="Akka.Actor.Props"/> instance configured to create the wrapped <see cref="DriverInstanceActor"/>.</returns>
public static Props Props(
IDriver driver,
@@ -420,7 +439,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
TimeSpan? shutdownTimeout = null,
int? maxWritesInFlight = null,
TimeSpan? queuedWriteBudget = null,
TimeSpan? initializeTimeout = null) =>
TimeSpan? initializeTimeout = null,
TimeSpan? writeTimeout = null) =>
Akka.Actor.Props.Create(() => new DriverInstanceActor(
driver,
reconnectInterval ?? DefaultReconnectInterval,
@@ -433,7 +453,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
shutdownTimeout,
maxWritesInFlight,
queuedWriteBudget,
initializeTimeout));
initializeTimeout,
writeTimeout));
/// <summary>
/// Returns true when the driver should boot in DEV-STUB mode based on host platform and
@@ -478,6 +499,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <see cref="DefaultQueuedWriteBudget"/> when null or non-positive.</param>
/// <param name="initializeTimeout">Pessimistic bound on one initialise attempt; defaults to
/// <see cref="DefaultInitializeTimeout"/> when null or non-positive.</param>
/// <param name="writeTimeout">Pessimistic bound on one write round-trip; defaults to
/// <see cref="DefaultWriteTimeout"/> when null or non-positive.</param>
public DriverInstanceActor(
IDriver driver,
TimeSpan reconnectInterval,
@@ -490,8 +513,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
TimeSpan? shutdownTimeout = null,
int? maxWritesInFlight = null,
TimeSpan? queuedWriteBudget = null,
TimeSpan? initializeTimeout = null)
TimeSpan? initializeTimeout = null,
TimeSpan? writeTimeout = null)
{
_writeTimeout = writeTimeout is { } wt && wt > TimeSpan.Zero ? wt : DefaultWriteTimeout;
_incarnation = Interlocked.Increment(ref s_initGenerationSource);
_maxWritesInFlight = maxWritesInFlight is > 0 ? maxWritesInFlight.Value : DefaultMaxWritesInFlight;
_queuedWriteBudget = queuedWriteBudget is { } qb && qb > TimeSpan.Zero ? qb : DefaultQueuedWriteBudget;
_initializeTimeout = initializeTimeout is { } it && it > TimeSpan.Zero ? it : DefaultInitializeTimeout;
@@ -1097,19 +1123,22 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
var self = Self;
var hostKey = ResolveHost(msg.TagId);
var invoker = _invoker;
_ = RunWriteAsync(writable, invoker, hostKey, msg, replyTo, self);
_ = RunWriteAsync(writable, invoker, hostKey, msg, replyTo, self, _writeTimeout, _incarnation, _log);
}
private static async Task RunWriteAsync(
IWritable writable, IDriverCapabilityInvoker invoker, string hostKey,
WriteAttribute msg, IActorRef replyTo, IActorRef self)
WriteAttribute msg, IActorRef replyTo, IActorRef self, TimeSpan writeTimeout, int incarnation,
ILoggingAdapter log)
{
// Off the actor context from here on: the first await yields, and nothing below reads actor state.
await Task.Yield();
var request = new[] { new WriteRequest(msg.TagId, msg.Value) };
// Bound the write so a hung backend can't hold a slot forever — retry stays
// off by default, but a stalled call still needs an answer.
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
// off by default, but a stalled call still needs an answer. Disposed below, except when the write is
// abandoned (the orphan may still be holding the token).
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
var abandoned = false;
try
{
// Route through the resilience invoker (Write pipeline: timeout, per-host breaker). Writes
@@ -1119,11 +1148,31 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// therefore authoritative regardless of any configured Write retry. Per-tag opt-in to retries via
// WriteIdempotentAttribute is the unshipped forward path (see WriteIdempotentAttribute). The outer
// cts is retained as a hard backstop above the (typically tighter) tier timeout.
var results = await invoker.ExecuteWriteAsync(
// PESSIMISTIC bound on top of the token-based cap: a driver that ignores its token would otherwise
// hold the slot and the tag forever, piling every later write to that tag up in the queue.
var attempt = invoker.ExecuteWriteAsync(
hostKey,
isIdempotent: false,
async ct => await writable.WriteAsync(request, ct),
cts.Token);
cts.Token).AsTask();
IReadOnlyList<WriteResult> results;
try
{
results = await attempt.WaitAsync(writeTimeout);
}
catch (TimeoutException)
{
abandoned = true;
log.Warning(
"DriverInstance write of {Tag} did not complete within {Bound} (the driver ignored its cancellation " +
"token); abandoned — its slot and tag are released, its eventual outcome is discarded",
msg.TagId, writeTimeout);
try { cts.Cancel(); } catch (ObjectDisposedException) { }
_ = attempt.ContinueWith(t => _ = t.Exception, TaskContinuationOptions.OnlyOnFaulted);
replyTo.Tell(new WriteAttributeResult(false,
$"StatusCode=0x800A0000 (BadTimeout: write did not complete within {writeTimeout.TotalSeconds:F0}s)"));
return;
}
if (results is { Count: 1 } && IsGoodStatus(results[0].StatusCode))
{
replyTo.Tell(new WriteAttributeResult(true, null));
@@ -1142,14 +1191,18 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
}
finally
{
self.Tell(new WriteCompleted(msg.TagId));
if (!abandoned) cts.Dispose();
self.Tell(new WriteCompleted(msg.TagId, incarnation));
}
}
/// <summary>Frees the write's slot and tag and drains the queue: the oldest entries whose tag is free are
/// launched (in arrival order) while slots remain; entries older than the budget are failed instead.</summary>
/// launched (in arrival order) while slots remain; entries older than the budget are failed instead. A
/// completion stamped with another incarnation (a write launched before a supervision restart) is ignored —
/// it never held a slot or tag of THIS incarnation.</summary>
private void OnWriteCompleted(WriteCompleted msg)
{
if (msg.Incarnation != _incarnation) return;
if (_writesInFlight > 0) _writesInFlight--;
_tagsInFlight.Remove(msg.TagId);
DrainWriteQueue();
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs
+38 -3
View File
@@ -23,12 +23,12 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
private static readonly TimeSpan WriteLatency = TimeSpan.FromMilliseconds(1200);
private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect(
IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null)
IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null, TimeSpan? writeTimeout = null)
{
var parent = CreateTestProbe();
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
driver, commsLossTimeout: TimeSpan.Zero, maxWritesInFlight: maxWritesInFlight,
queuedWriteBudget: queuedWriteBudget));
queuedWriteBudget: queuedWriteBudget, writeTimeout: writeTimeout));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(m => m.Connected, PresenceBudget);
return (actor, parent);
@@ -163,6 +163,38 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
driver.EnteredRefs.ShouldBe(new[] { "tag-0" }, "the stale write must never reach the device");
}
/// <summary>
/// Fix round 2: a token-ignoring hung write held its tag forever, so every later write to that tag piled up
/// in the queue. The write is bounded pessimistically; on overrun it answers BadTimeout and releases its
/// slot and tag, and a later write to the same tag reaches the device.
/// </summary>
[Fact]
public async Task A_hung_token_ignoring_write_is_abandoned_and_releases_its_tag_for_a_later_write()
{
var driver = new SlowWritableDriver { HoldWrites = true, IgnoreCancellation = true };
var (actor, _) = Connect(driver, maxWritesInFlight: 1, writeTimeout: TimeSpan.FromMilliseconds(300));
var hung = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget);
AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget);
var hungResult = await hung;
hungResult.Success.ShouldBeFalse();
hungResult.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout");
driver.MaxConcurrentWrites.ShouldBe(1);
// The tag (and the single slot) are free again: a later tag-a write reaches the device even though the
// abandoned one is still inside it.
var later = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
new DriverInstanceActor.WriteAttribute("tag-a", 2), PresenceBudget);
AwaitCondition(() => driver.WritesEntered == 2, PresenceBudget);
driver.EnteredRefs.ShouldBe(new[] { "tag-a", "tag-a" });
driver.MaxConcurrentWrites.ShouldBe(2, "positive control: the abandoned write really is still wedged inside the driver");
driver.ReleaseWrites();
(await later).Success.ShouldBeTrue();
}
[Fact]
public async Task Queued_writes_fail_fast_when_the_driver_leaves_Connected()
{
@@ -201,6 +233,8 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
public TimeSpan Latency { get; init; }
public bool HoldWrites { get; init; }
/// <summary>When true a held write ignores its cancellation token — a wedged driver.</summary>
public bool IgnoreCancellation { get; init; }
public IReadOnlyList<WriteRequest> Writes => _writes.ToArray();
public IReadOnlyList<string> EnteredRefs => _enteredRefs.ToArray();
@@ -227,7 +261,8 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
Interlocked.CompareExchange(ref _maxInside, inside, seen);
try
{
if (HoldWrites) await _release.Task.WaitAsync(cancellationToken);
if (HoldWrites && IgnoreCancellation) await _release.Task;
else if (HoldWrites) await _release.Task.WaitAsync(cancellationToken);
else if (Latency > TimeSpan.Zero) await Task.Delay(Latency, cancellationToken);
foreach (var w in writes) _writes.Enqueue(w);
return writes.Select(_ => new WriteResult(OpcUaStatus.Good)).ToList();