From 75be6fe12b195c7d3a21768238b88de7c7d589d2 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 9 Nov 2026 15:18:37 -0500 Subject: [PATCH] fix(runtime): derive the write cap from the configured write timeout (lmxopcua#774) DriverInstanceActor.RunWriteAsync cancelled every write with a hard-coded, undilated 5 s CancellationTokenSource, so writeTimeout / DefaultWriteTimeout (then 10 s) only ever served as the pessimistic abandon bound: a configured bound above 5 s never took effect. The write's token is now cancelled at the configured write timeout; a write that ignores its token is abandoned WriteAbandonGrace later (half the timeout, at most 2 s) so a token-honouring driver still gets to answer the cancellation itself. Both bounds run on an injectable TimeProvider (Props writeClock, System by default). DefaultWriteTimeout is now 5 s (cancel 5 s, abandon 7 s), not 10 s: the host's write Ask is 8 s, and a write allowed to run past it would already have been reported as a timeout and its node reverted while the device could still take the value - the exact failure the off-mailbox write work fixed. Production behaviour is therefore unchanged for the cancel (5 s) and tighter for a token-ignoring driver (abandoned at 7 s instead of 10 s, i.e. inside the Ask). DriverHostActor's Ask comment updated to match. Tests (DriverInstanceActorConcurrentWriteTests, manual clock copied from Ab.Transport.Tests into Runtime.Tests/Harness/ManualTimeProvider): - a write held past 6 s under a 10 s write timeout succeeds; - positive control: at 10 s the token is cancelled ("write timeout"); - a token-ignoring write is still pending at 10 s and abandoned (BadTimeout) after the grace. Mutation checks: a fixed 5 s token on the same clock fails the first test; dropping the grace fails the third (and the second, where the abandon then races the cancellation). 342a825d re-check: its held-write tests were written around the undilated 5 s cap. Connect() now defaults the write timeout to the DILATED presence budget, so a held, explicitly-released write no longer races any undilated deadline; tests about the bound pass their own. AwaitWritesEntered keeps its 20 ms interval (doc updated). Runtime.Tests: 946 passed / 3 skipped / 0 failed at AKKA_TEST_TIMEFACTOR 3 and 1. Solution Release build: 0 warnings. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV --- .../Drivers/DriverHostActor.cs | 5 +- .../Drivers/DriverInstanceActor.cs | 71 ++++++++---- ...DriverInstanceActorConcurrentWriteTests.cs | 104 ++++++++++++++++- .../Harness/ManualTimeProvider.cs | 106 ++++++++++++++++++ 4 files changed, 259 insertions(+), 27 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ManualTimeProvider.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 49abd04dc..f8d152bf3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -2237,8 +2237,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS return; } - // Ask the child a bounded write — DriverInstanceActor.HandleWriteAsync already bounds the backend - // call to 5s, so the 8s Ask is a safety net for an actor that never replies. Capture Sender before + // Ask the child a bounded write — DriverInstanceActor bounds the backend call by its write timeout + // (DefaultWriteTimeout 5 s; a token-ignoring write is abandoned at 7 s), so the 8s Ask is a safety net + // for an actor that never replies. Capture Sender before // the continuation: it runs off the actor thread, where raw Sender is unsafe to read. var replyTo = Sender; entry.Actor.Ask( 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 f5ed46709..d06efe9a6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -229,11 +229,30 @@ 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. + /// Bound on one write round-trip. The write's cancellation token is cancelled when it elapses (the write + /// answers "write timeout"); a driver whose ignores its token is + /// abandoned later (the write answers BadTimeout and frees its slot + /// and tag), so it cannot hold its slot AND its tag forever, piling every later write to that tag up in the + /// queue. /// - public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(10); + /// + /// 5 s, so both bounds (5 s cancel, 7 s abandon) answer inside the host's 8 s write Ask + /// (DriverHostActor): a write that outlives the Ask has already been reported as a timeout and its + /// node reverted upstream, so letting it run on would let the device take a value the client was told + /// failed. Until lmxopcua#774 the cancel was a hard-coded 5 s that ignored this value, and the default was + /// 10 s — a configured bound above 5 s never took effect. + /// + public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(5); + + /// + /// How long after a write's token is cancelled () a write that has not + /// returned is abandoned as token-ignoring: half the write timeout, at most 2 s. Long enough for a driver + /// that honours its token to unwind and answer the cancellation itself. + /// + /// The write timeout in force. + /// The grace between cancelling the write's token and abandoning the write. + internal static TimeSpan WriteAbandonGrace(TimeSpan writeTimeout) => + TimeSpan.FromTicks(Math.Min(writeTimeout.Ticks / 2, TimeSpan.FromSeconds(2).Ticks)); /// Self-sent by a write task once it has replied, releasing its slot and its tag. /// The tag the write held. @@ -289,6 +308,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// See . private readonly TimeSpan _writeTimeout; + /// The clock the write bounds run on — outside tests. + private readonly TimeProvider _writeClock; + /// 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; @@ -434,8 +456,11 @@ 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 + /// Optional bound on one write round-trip (its token is cancelled when it elapses; + /// a token-ignoring write is abandoned later); defaults to /// . + /// Optional clock for the write bounds; defaults to . + /// Exists so a test can step past a bound without waiting for it. /// A instance configured to create the wrapped . public static Props Props( IDriver driver, @@ -450,7 +475,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null, TimeSpan? initializeTimeout = null, - TimeSpan? writeTimeout = null) => + TimeSpan? writeTimeout = null, + TimeProvider? writeClock = null) => Akka.Actor.Props.Create(() => new DriverInstanceActor( driver, reconnectInterval ?? DefaultReconnectInterval, @@ -464,7 +490,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers maxWritesInFlight, queuedWriteBudget, initializeTimeout, - writeTimeout)); + writeTimeout, + writeClock)); /// /// Returns true when the driver should boot in DEV-STUB mode based on host platform and @@ -509,8 +536,9 @@ 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 + /// Bound on one write round-trip; defaults to /// when null or non-positive. + /// Clock for the write bounds; defaults to . public DriverInstanceActor( IDriver driver, TimeSpan reconnectInterval, @@ -524,9 +552,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null, TimeSpan? initializeTimeout = null, - TimeSpan? writeTimeout = null) + TimeSpan? writeTimeout = null, + TimeProvider? writeClock = null) { _writeTimeout = writeTimeout is { } wt && wt > TimeSpan.Zero ? wt : DefaultWriteTimeout; + _writeClock = writeClock ?? TimeProvider.System; _incarnation = Interlocked.Increment(ref s_initGenerationSource); _maxWritesInFlight = maxWritesInFlight is > 0 ? maxWritesInFlight.Value : DefaultMaxWritesInFlight; _queuedWriteBudget = queuedWriteBudget is { } qb && qb > TimeSpan.Zero ? qb : DefaultQueuedWriteBudget; @@ -1139,21 +1169,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, _writeTimeout, _incarnation, _log); + _ = RunWriteAsync(writable, invoker, hostKey, msg, replyTo, self, _writeTimeout, _writeClock, _incarnation, _log); } private static async Task RunWriteAsync( IWritable writable, IDriverCapabilityInvoker invoker, string hostKey, - WriteAttribute msg, IActorRef replyTo, IActorRef self, TimeSpan writeTimeout, int incarnation, - ILoggingAdapter log) + WriteAttribute msg, IActorRef replyTo, IActorRef self, TimeSpan writeTimeout, TimeProvider clock, + 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. Disposed below, except when the write is - // abandoned (the orphan may still be holding the token). - var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + // off by default, but a stalled call still needs an answer. The bound is the configured write timeout + // (lmxopcua#774: it used to be a hard-coded 5 s, so a longer configured bound never took effect). + // Disposed below, except when the write is abandoned (the orphan may still be holding the token). + var cts = new CancellationTokenSource(writeTimeout, clock); var abandoned = false; try { @@ -1164,7 +1195,7 @@ 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. - // PESSIMISTIC bound on top of the token-based cap: a driver that ignores its token would otherwise + // PESSIMISTIC bound a grace after 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, @@ -1174,15 +1205,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers IReadOnlyList results; try { - results = await attempt.WaitAsync(writeTimeout); + results = await attempt.WaitAsync(writeTimeout + WriteAbandonGrace(writeTimeout), clock); } 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); + "DriverInstance write of {Tag} did not complete within {Bound} and ignored its cancellation token " + + "for {Grace} more; abandoned — its slot and tag are released, its eventual outcome is discarded", + msg.TagId, writeTimeout, WriteAbandonGrace(writeTimeout)); try { cts.Cancel(); } catch (ObjectDisposedException) { } _ = attempt.ContinueWith(t => _ = t.Exception, TaskContinuationOptions.OnlyOnFaulted); replyTo.Tell(new WriteAttributeResult(false, 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 588bb431a..87f082a47 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 @@ -26,21 +26,31 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa /// Polls for writes inside the driver with an EXPLICIT short interval. Akka's /// AwaitCondition without one polls every max / 10 of the DILATED budget — 1.5 s locally, 4.5 s /// under CI's AKKA_TEST_TIMEFACTOR=3. The write under test is already held inside the driver by then, - /// and DriverInstanceActor.RunWriteAsync bounds it with a fixed (undilated) 5 s token, so one missed - /// first poll plus the test's own settle window cancelled it ("write timeout") before + /// and it was bounded by DriverInstanceActor.RunWriteAsync's fixed (undilated) 5 s token, so one + /// missed first poll plus the test's own settle window cancelled it ("write timeout") before /// ReleaseWrites(): A_queued_write_older_than_the_budget… failed every CI run that hit it - /// (hosted runs 2215 and 2260), and fails locally at timefactor 3. + /// (hosted runs 2215 and 2260), and failed locally at timefactor 3. Since lmxopcua#774 that token follows + /// the actor's write timeout, and sets it to the dilated presence budget, so a held + /// write no longer races an undilated bound at all; the short interval stays to keep the tests fast. /// private void AwaitWritesEntered(SlowWritableDriver driver, int count) => AwaitCondition(() => driver.WritesEntered == count, PresenceBudget, TimeSpan.FromMilliseconds(20)); + /// + /// Spawns and connects the actor. Unless a test passes its own , the write + /// bound is the DILATED presence budget: these tests hold writes inside the driver and release them + /// explicitly, and are not about the bound — the production 5 s default would be an undilated deadline + /// racing every held write under CI's timefactor (the 342a825d failure mode). + /// private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect( - IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null, TimeSpan? writeTimeout = null) + IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null, TimeSpan? writeTimeout = null, + TimeProvider? writeClock = null) { var parent = CreateTestProbe(); var actor = parent.ChildActorOf(DriverInstanceActor.Props( driver, commsLossTimeout: TimeSpan.Zero, maxWritesInFlight: maxWritesInFlight, - queuedWriteBudget: queuedWriteBudget, writeTimeout: writeTimeout)); + queuedWriteBudget: queuedWriteBudget, writeTimeout: writeTimeout ?? Dilated(PresenceBudget), + writeClock: writeClock)); actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); parent.ExpectMsg(m => m.Connected, PresenceBudget); return (actor, parent); @@ -213,6 +223,90 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa driver.Writes.Select(w => w.Value).ShouldBe(new object[] { 2 }, "only the later write reached the device's commit"); } + /// + /// lmxopcua#774: the write's cancellation token was a hard-coded 5 s, so a configured write timeout above + /// 5 s never took effect. On a manual clock: a write held past 6 s under a 10 s write timeout still + /// succeeds when the device answers. (Mutation-checked: a fixed 5 s token on the same clock fails this + /// with "write timeout".) + /// + [Fact] + public async Task A_write_longer_than_5s_but_within_the_configured_write_timeout_succeeds() + { + var clock = new ManualTimeProvider(); + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, _) = Connect(driver, writeTimeout: TimeSpan.FromSeconds(10), writeClock: clock); + + var write = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget); + AwaitWritesEntered(driver, 1); + AwaitWriteBoundsArmed(clock); + + clock.Advance(TimeSpan.FromSeconds(6)); + await Task.Delay(100); + write.IsCompleted.ShouldBeFalse("the write was cut off before its configured 10 s bound"); + driver.ReleaseWrites(); + + var result = await write; + result.Success.ShouldBeTrue(result.Reason); + driver.Writes.Select(w => w.FullReference).ShouldBe(new[] { "tag-a" }); + } + + /// + /// Positive control for the test above: the configured write timeout DOES bound the write — at 10 s its + /// token is cancelled and a token-honouring driver answers "write timeout" (not the abandon path). + /// + [Fact] + public async Task The_configured_write_timeout_cancels_the_write_when_it_elapses() + { + var clock = new ManualTimeProvider(); + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, _) = Connect(driver, writeTimeout: TimeSpan.FromSeconds(10), writeClock: clock); + + var write = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget); + AwaitWritesEntered(driver, 1); + AwaitWriteBoundsArmed(clock); + + clock.Advance(TimeSpan.FromSeconds(10)); + var result = await write; + result.Success.ShouldBeFalse(); + result.Reason.ShouldBe("write timeout"); + driver.Writes.ShouldBeEmpty(); + } + + /// + /// A token-IGNORING write is abandoned after its token + /// is cancelled, not at the same instant (a token-honouring driver gets the grace to answer itself). + /// + [Fact] + public async Task A_token_ignoring_write_is_abandoned_a_grace_after_the_configured_write_timeout() + { + var clock = new ManualTimeProvider(); + var driver = new SlowWritableDriver { HangFirstWriteIgnoringCancellation = true }; + var (actor, _) = Connect(driver, writeTimeout: TimeSpan.FromSeconds(10), writeClock: clock); + + var write = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget); + AwaitWritesEntered(driver, 1); + AwaitWriteBoundsArmed(clock); + + clock.Advance(TimeSpan.FromSeconds(10)); + await Task.Delay(100); + write.IsCompleted.ShouldBeFalse("abandoned at the cancel instant — no grace for the driver to answer"); + + clock.Advance(DriverInstanceActor.WriteAbandonGrace(TimeSpan.FromSeconds(10))); + var result = await write; + result.Success.ShouldBeFalse(); + result.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout"); + result.Reason!.ShouldContain("within 10000 ms"); + } + + /// Waits until the write task has armed BOTH of its bounds on (the token + /// cancel and the abandon), so an Advance cannot run ahead of them. Counts the timers rather than matching their + /// due times, so a bound armed at the WRONG time fails the test's own assertion instead of this wait. + private void AwaitWriteBoundsArmed(ManualTimeProvider clock) => + AwaitCondition(() => clock.ArmedTimers == 2, PresenceBudget, TimeSpan.FromMilliseconds(20)); + [Fact] public async Task Queued_writes_fail_fast_when_the_driver_leaves_Connected() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ManualTimeProvider.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ManualTimeProvider.cs new file mode 100644 index 000000000..4b0c25409 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ManualTimeProvider.cs @@ -0,0 +1,106 @@ +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +/// +/// Deterministic : time moves only on , which fires every +/// timer falling due on the way, in due order, synchronously on the calling thread. +/// (Microsoft.Extensions.TimeProvider.Testing is not pinned in this repo; a copy of the Ab.Transport.Tests fake.) +/// +internal sealed class ManualTimeProvider : TimeProvider +{ + private readonly object _lock = new(); + private readonly List _timers = []; + private DateTimeOffset _now; + + public ManualTimeProvider() => Start = _now = new DateTimeOffset(2026, 10, 7, 0, 0, 0, TimeSpan.Zero); + + public DateTimeOffset Start { get; } + + /// Steps the wall clock () without moving monotonic time + /// () or any timer, as an NTP or operator clock change does. + public TimeSpan WallClockOffset { get; set; } + + public override DateTimeOffset GetUtcNow() { lock (_lock) return _now + WallClockOffset; } + + public override long GetTimestamp() { lock (_lock) return _now.UtcTicks; } + + public override long TimestampFrequency => TimeSpan.TicksPerSecond; + + public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period) + { + var t = new ManualTimer(this, callback, state); + lock (_lock) _timers.Add(t); + t.Change(dueTime, period); + return t; + } + + /// Armed timers (not disposed, with a due time). + public int ArmedTimers { get { lock (_lock) return _timers.Count(t => t.Due is not null); } } + + public bool HasTimerDueAt(DateTimeOffset at) { lock (_lock) return _timers.Any(t => t.Due == at); } + + /// Waits (in real time) until some code under test has armed a timer due at , + /// i.e. is genuinely parked on the fake clock rather than merely not yet scheduled. + public async Task WaitForTimerDueAtAsync(DateTimeOffset at, TimeSpan timeout) + { + var giveUp = DateTime.UtcNow + timeout; + while (!HasTimerDueAt(at)) + { + if (DateTime.UtcNow > giveUp) throw new TimeoutException($"No timer armed for {at:O} (now {GetUtcNow():O})."); + await Task.Delay(5); + } + } + + public void Advance(TimeSpan by) + { + DateTimeOffset target; + lock (_lock) target = _now + by; + while (true) + { + ManualTimer? next; + lock (_lock) + { + next = _timers.Where(t => t.Due <= target).MinBy(t => t.Due); + if (next is null) { _now = target; return; } + _now = next.Due!.Value; + next.Due = next.Period is { } p ? _now + p : null; + } + next.Fire(); + } + } + + private sealed class ManualTimer(ManualTimeProvider owner, TimerCallback callback, object? state) : ITimer + { + public DateTimeOffset? Due; // guarded by owner._lock + public TimeSpan? Period; // guarded by owner._lock + private bool _disposed; + + public void Fire() => callback(state); + + public bool Change(TimeSpan dueTime, TimeSpan period) + { + lock (owner._lock) + { + if (_disposed) return false; + Due = dueTime == Timeout.InfiniteTimeSpan ? null : owner._now + dueTime; + Period = period == Timeout.InfiniteTimeSpan || period == TimeSpan.Zero ? null : period; + return true; + } + } + + public void Dispose() + { + lock (owner._lock) + { + _disposed = true; + Due = null; + owner._timers.Remove(this); + } + } + + public ValueTask DisposeAsync() + { + Dispose(); + return ValueTask.CompletedTask; + } + } +}