From 7ae74ef147b3a5004a1c85dbf474551b346aafb9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 10:43:45 -0400 Subject: [PATCH] fix(runtime): pessimistic write bound, incarnation-stamped write completions (fix round 2) - 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 --- .../Drivers/DriverInstanceActor.cs | 77 ++++++++++++++++--- ...DriverInstanceActorConcurrentWriteTests.cs | 41 +++++++++- 2 files changed, 103 insertions(+), 15 deletions(-) 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 e1924364..60e989ce 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -218,8 +218,18 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// public static readonly TimeSpan DefaultInitializeTimeout = TimeSpan.FromSeconds(30); + /// + /// Pessimistic bound on one write round-trip, on top of the 5 s token-based write cap. A driver whose + /// 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. + /// + public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(10); + /// Self-sent by a write task once it has replied, releasing its slot and its tag. - private sealed record WriteCompleted(string TagId); + /// The tag the write held. + /// 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. + private sealed record WriteCompleted(string TagId, int Incarnation); private readonly IDriver _driver; @@ -266,6 +276,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// See . private readonly TimeSpan _initializeTimeout; + /// See . + private readonly TimeSpan _writeTimeout; + + /// 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). + private readonly int _incarnation; + /// /// Set by so the it triggers skips the dispose. A /// supervision restart recreates the actor object around the SAME (the Props closure @@ -407,6 +424,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// BadTimeout; defaults to . /// Optional pessimistic bound on one initialise attempt; defaults to /// . + /// Optional pessimistic bound on one write round-trip; defaults to + /// . /// A instance configured to create the wrapped . 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)); /// /// 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 /// when null or non-positive. /// Pessimistic bound on one initialise attempt; defaults to /// when null or non-positive. + /// Pessimistic bound on one write round-trip; defaults to + /// when null or non-positive. 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 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)); } } /// 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. + /// 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. private void OnWriteCompleted(WriteCompleted msg) { + if (msg.Incarnation != _incarnation) return; if (_writesInFlight > 0) _writesInFlight--; _tagsInFlight.Remove(msg.TagId); DrainWriteQueue(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs index fc2f041f..f6f99002 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs @@ -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(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"); } + /// + /// 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. + /// + [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( + 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( + 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; } + /// When true a held write ignores its cancellation token — a wedged driver. + public bool IgnoreCancellation { get; init; } public IReadOnlyList Writes => _writes.ToArray(); public IReadOnlyList 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();