diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs index f7b5ef56..71169d81 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Clients/FleetDiagnosticsClient.cs @@ -42,9 +42,11 @@ public sealed class FleetDiagnosticsClient : IFleetDiagnosticsClient /// public async Task GetDiagnosticsAsync(NodeId nodeId, CancellationToken ct) { - var selection = _system.ActorSelection($"akka.tcp://{_systemName}@{nodeId.Value}/user/driver-host"); try { + // Inside the try: a malformed node value makes ActorSelection throw, and that must degrade to + // "unavailable" like a dead node does, not surface through the page's error boundary. + var selection = _system.ActorSelection($"akka.tcp://{_systemName}@{nodeId.Value}/user/driver-host"); using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct); linked.CancelAfter(AskTimeout); return await selection.Ask( diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/FleetDiagnostics.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/FleetDiagnostics.razor index b2171f3a..1b958048 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/FleetDiagnostics.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/FleetDiagnostics.razor @@ -120,7 +120,7 @@ else State Connected devices Faulted devices - Last change + Last health publish diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs index 3d6af8fe..4d542e7a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Telemetry/TelemetryDialSupervisor.cs @@ -252,6 +252,9 @@ public sealed class TelemetryDialSupervisor : ReceiveActor, IWithTimers return; } + // The stream is over; a still-armed establishment deadline for it must not fire into the reconnect. + Timers.Cancel(DeadlineTimerKey(msg.NodeId)); + if (dialer.Connected) { dialer.Connected = false; 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 3eeaa913..e1924364 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using Akka.Actor; using Akka.Event; using ZB.MOM.WW.OtOpcUa.Cluster; @@ -203,12 +204,22 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// public const int DefaultMaxWritesInFlight = 4; - /// Self-sent by a write task once it has replied, releasing its in-flight slot. - private sealed class WriteCompleted - { - public static readonly WriteCompleted Instance = new(); - private WriteCompleted() { } - } + /// + /// How long a queued write may wait for a slot before it is failed BadTimeout instead of being + /// written. The host's write Ask is 8 s: a write older than that has already been reported as a timeout + /// (and its node reverted) upstream, so writing it late would push a stale value to the device. + /// + public static readonly TimeSpan DefaultQueuedWriteBudget = TimeSpan.FromSeconds(8); + + /// + /// Pessimistic bound on one attempt. The invoker's tier timeout is + /// token-only (optimistic), so a driver whose initialise ignores its token would otherwise hold the + /// single-flight gate forever and wedge the reconnect loop. 2× the longest tier (Tier C: 15 s). + /// + public static readonly TimeSpan DefaultInitializeTimeout = TimeSpan.FromSeconds(30); + + /// Self-sent by a write task once it has replied, releasing its slot and its tag. + private sealed record WriteCompleted(string TagId); private readonly IDriver _driver; @@ -240,9 +251,20 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// the self-sent ). private int _writesInFlight; - /// Writes waiting for an in-flight slot, in arrival order. Failed fast on every exit from - /// Connected, so a queued write never reaches a driver that is reconnecting. - private readonly Queue<(WriteAttribute Write, IActorRef ReplyTo)> _pendingWrites = new(); + /// Writes waiting for a slot (or for their tag), in arrival order, stamped with arrival time. + /// Failed fast on every exit from Connected, so a queued write never reaches a driver that is + /// reconnecting; failed BadTimeout at dequeue once older than . + private readonly LinkedList<(WriteAttribute Write, IActorRef ReplyTo, DateTime EnqueuedUtc)> _pendingWrites = new(); + + /// Tags with a write currently inside the driver. A second write to the same tag waits in the + /// queue behind it — two concurrent writes v1, v2 to one tag could otherwise land as v1. + private readonly HashSet _tagsInFlight = new(StringComparer.Ordinal); + + /// See . + private readonly TimeSpan _queuedWriteBudget; + + /// See . + private readonly TimeSpan _initializeTimeout; /// /// Set by so the it triggers skips the dispose. A @@ -267,6 +289,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// on the actor thread, so no lock is needed. private int _initGeneration; + /// Process-wide source of attempt generations. A supervision restart recreates the actor object + /// (fresh fields) while a prior incarnation's attempt may still be running on the same driver; a per-actor + /// counter restarting at 0 would let that late result collide with the new incarnation's first attempt. + private static int s_initGenerationSource; + /// /// True while an attempt is running (set when it is launched, cleared by its /// piped /). The Reconnecting retry @@ -376,6 +403,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// hung shutdown is abandoned without waiting 5 s. /// Optional cap on concurrent writes against the driver; defaults to /// . Writes beyond it queue in arrival order. + /// Optional bound on how long a queued write may wait before it is failed + /// BadTimeout; defaults to . + /// Optional pessimistic bound on one initialise attempt; defaults to + /// . /// A instance configured to create the wrapped . public static Props Props( IDriver driver, @@ -387,7 +418,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers TimeSpan? healthPollInterval = null, TimeSpan? commsLossTimeout = null, TimeSpan? shutdownTimeout = null, - int? maxWritesInFlight = null) => + int? maxWritesInFlight = null, + TimeSpan? queuedWriteBudget = null, + TimeSpan? initializeTimeout = null) => Akka.Actor.Props.Create(() => new DriverInstanceActor( driver, reconnectInterval ?? DefaultReconnectInterval, @@ -398,7 +431,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers healthPollInterval, commsLossTimeout, shutdownTimeout, - maxWritesInFlight)); + maxWritesInFlight, + queuedWriteBudget, + initializeTimeout)); /// /// Returns true when the driver should boot in DEV-STUB mode based on host platform and @@ -439,6 +474,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// when null. /// Cap on concurrent writes; defaults to /// when null or non-positive. + /// Bound on a queued write's wait; defaults to + /// when null or non-positive. + /// Pessimistic bound on one initialise attempt; defaults to + /// when null or non-positive. public DriverInstanceActor( IDriver driver, TimeSpan reconnectInterval, @@ -449,9 +488,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers TimeSpan? healthPollInterval = null, TimeSpan? commsLossTimeout = null, TimeSpan? shutdownTimeout = null, - int? maxWritesInFlight = null) + int? maxWritesInFlight = null, + TimeSpan? queuedWriteBudget = null, + TimeSpan? initializeTimeout = null) { _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; _driver = driver; _invoker = invoker ?? NullDriverCapabilityInvoker.Instance; _driverInstanceId = driver.DriverInstanceId; @@ -542,7 +585,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Receive(_ => Sender.Tell(new WriteAttributeResult(false, "driver not connected"))); // A write launched while Connected can finish after the transition out of it; its slot still frees. - Receive(_ => OnWriteCompleted()); + Receive(OnWriteCompleted); // An ack arriving while still connecting can't reach the upstream alarm system; drop it (the ack is // fire-and-forget — no reply to surface — and the OPC UA condition state already committed locally). Receive(_ => @@ -618,7 +661,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // Synchronous Receive: the device call runs off the mailbox (see StartOrQueueWrite), so data changes // and further writes are never head-of-line blocked behind a slow device. Receive(msg => StartOrQueueWrite(msg, Sender)); - Receive(_ => OnWriteCompleted()); + Receive(OnWriteCompleted); ReceiveAsync(HandleAcknowledgeAsync); ReceiveAsync(HandleSubscribeAsync); ReceiveAsync(_ => UnsubscribeAsync()); @@ -846,7 +889,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // timeout on an inbound write to a transiently-down driver). Synchronous Receive. Receive(_ => Sender.Tell(new WriteAttributeResult(false, "driver not connected"))); - Receive(_ => OnWriteCompleted()); + Receive(OnWriteCompleted); // An ack arriving while reconnecting can't reach the upstream alarm system; drop it (fire-and-forget, // no reply — the OPC UA condition state already committed locally on the Part 9 ack). Receive(_ => @@ -925,7 +968,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } _initInFlight = true; - var generation = ++_initGeneration; + var generation = Interlocked.Increment(ref s_initGenerationSource); + _initGeneration = generation; var initCts = new CancellationTokenSource(); _initCts = initCts; var self = Self; @@ -934,6 +978,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers var driver = _driver; var invoker = _invoker; var shutdownTimeout = _shutdownTimeout; + var initializeTimeout = _initializeTimeout; _ = Task.Run(async () => { try @@ -973,11 +1018,30 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // 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( + // PESSIMISTIC bound on top of the invoker's token-only timeout: a driver that ignores its token + // would otherwise hold the single-flight gate forever. An overrun is abandoned (the task keeps + // running, its eventual outcome is observed and discarded), reported as a failure so the retry + // loop continues, and its late Tell is dropped by the generation guard anyway. + var attempt = invoker.ExecuteAsync( DriverCapability.Initialize, driverInstanceId, async ct => await driver.InitializeAsync(driverConfigJson, ct), - initCts.Token); + initCts.Token).AsTask(); + try + { + await attempt.WaitAsync(initializeTimeout); + } + catch (TimeoutException) + { + log.Warning( + "DriverInstance {Id}: initialise attempt {Gen} did not complete within {Bound} (the driver ignored " + + "its cancellation token); attempt abandoned, the retry loop continues", + driverInstanceId, generation, initializeTimeout); + try { initCts.Cancel(); } catch (ObjectDisposedException) { } + _ = attempt.ContinueWith(t => _ = t.Exception, TaskContinuationOptions.OnlyOnFaulted); + self.Tell(new InitializeFailed($"initialise did not complete within {initializeTimeout}", generation)); + return; + } self.Tell(new InitializeSucceeded(generation)); } catch (Exception ex) @@ -986,7 +1050,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } finally { - initCts.Dispose(); + // The CTS outlives an abandoned attempt (it may still be inside the driver); dispose only once + // nothing can cancel it any more — the actor nulls _initCts when it honours the result. + if (!initCts.IsCancellationRequested) initCts.Dispose(); } }); } @@ -1013,13 +1079,19 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers return; } - if (_writesInFlight >= _maxWritesInFlight) + if (_writesInFlight >= _maxWritesInFlight || _tagsInFlight.Contains(msg.TagId)) { - _pendingWrites.Enqueue((msg, replyTo)); + _pendingWrites.AddLast((msg, replyTo, DateTime.UtcNow)); return; } + LaunchWrite(writable, msg, replyTo); + } + + private void LaunchWrite(IWritable writable, WriteAttribute msg, IActorRef replyTo) + { _writesInFlight++; + _tagsInFlight.Add(msg.TagId); // Everything the task needs is captured HERE, on the actor thread — ResolveHost reads the driver, // and Self must not be touched off-context. var self = Self; @@ -1070,18 +1142,43 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } finally { - self.Tell(WriteCompleted.Instance); + self.Tell(new WriteCompleted(msg.TagId)); } } - /// Frees a write slot and, while Connected, starts the next queued write. - private void OnWriteCompleted() + /// 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. + private void OnWriteCompleted(WriteCompleted msg) { if (_writesInFlight > 0) _writesInFlight--; - if (_pendingWrites.Count > 0 && _driver is IWritable) + _tagsInFlight.Remove(msg.TagId); + DrainWriteQueue(); + } + + private void DrainWriteQueue() + { + if (_driver is not IWritable writable) return; + var now = DateTime.UtcNow; + var node = _pendingWrites.First; + while (node is not null && _writesInFlight < _maxWritesInFlight) { - var (next, replyTo) = _pendingWrites.Dequeue(); - StartOrQueueWrite(next, replyTo); + var next = node.Next; + var (write, replyTo, enqueuedUtc) = node.Value; + if (now - enqueuedUtc > _queuedWriteBudget) + { + // Already reported as a timeout upstream (the host's Ask is the budget) and the node reverted — + // writing it now would push a stale value to the device. BadTimeout, never written. + _pendingWrites.Remove(node); + replyTo.Tell(new WriteAttributeResult(false, + $"StatusCode=0x800A0000 (BadTimeout: queued for {(now - enqueuedUtc).TotalSeconds:F1}s, over the {_queuedWriteBudget.TotalSeconds:F0}s write budget)")); + } + else if (!_tagsInFlight.Contains(write.TagId)) + { + _pendingWrites.Remove(node); + LaunchWrite(writable, write, replyTo); + } + + node = next; } } @@ -1090,10 +1187,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// real outcome. private void FailQueuedWrites(string reason) { - while (_pendingWrites.Count > 0) + while (_pendingWrites.First is { } node) { - var (_, replyTo) = _pendingWrites.Dequeue(); - replyTo.Tell(new WriteAttributeResult(false, reason)); + _pendingWrites.RemoveFirst(); + node.Value.ReplyTo.Tell(new WriteAttributeResult(false, reason)); } } @@ -1753,9 +1850,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers 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. + // BOUNDED by ONE shared budget across shutdown + dispose. 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 is + // abandoned, and the dispose gets whatever is left of the same budget (never a second full one). + var budget = Stopwatch.StartNew(); using (var cts = new CancellationTokenSource(_shutdownTimeout)) { try { _driver.ShutdownAsync(cts.Token).WaitAsync(_shutdownTimeout).GetAwaiter().GetResult(); } @@ -1774,23 +1873,34 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // driver's compiled-script AssemblyLoadContexts and an OpcUaClient driver's LoggerFactory / gate / CTS // leaked once per respawn (02/L1). Bounded like the shutdown, and runs even after a timed-out shutdown — // leaking the object is strictly worse than disposing one whose shutdown is still wedged. - if (!_restarting) DisposeDriver(); + if (!_restarting) + { + var remaining = _shutdownTimeout - budget.Elapsed; + DisposeDriver(remaining > MinDisposeBudget ? remaining : MinDisposeBudget); + } OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1, new KeyValuePair("event", "stop"), new KeyValuePair("driver_type", _driver.DriverType)); } + /// Floor on the dispose's share of the PostStop budget once a timed-out shutdown has used it up — + /// enough for an idempotent dispose to run, not enough to double the stop's thread time. + private static readonly TimeSpan MinDisposeBudget = TimeSpan.FromMilliseconds(250); + /// Final teardown of the wrapped driver: preferred, else - /// , never both. A driver implementing neither needs nothing here. - private void DisposeDriver() + /// , never both. A driver implementing neither needs nothing here. Runs even after + /// a timed-out shutdown, so a driver's dispose must be safe CONCURRENTLY with its own still-running + /// . + /// The remaining PostStop budget available to an async dispose. + private void DisposeDriver(TimeSpan bound) { try { switch (_driver) { case IAsyncDisposable asyncDisposable: - asyncDisposable.DisposeAsync().AsTask().WaitAsync(_shutdownTimeout).GetAwaiter().GetResult(); + asyncDisposable.DisposeAsync().AsTask().WaitAsync(bound).GetAwaiter().GetResult(); break; case IDisposable disposable: disposable.Dispose(); @@ -1807,7 +1917,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers catch (TimeoutException) { _log.Warning("DriverInstance {Id}: DisposeAsync did not complete within {Bound} on PostStop; abandoning it", - _driverInstanceId, _shutdownTimeout); + _driverInstanceId, bound); } catch (Exception ex) { 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 b4bf5683..fc2f041f 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 @@ -22,11 +22,13 @@ 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) + private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect( + IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null) { var parent = CreateTestProbe(); var actor = parent.ChildActorOf(DriverInstanceActor.Props( - driver, commsLossTimeout: TimeSpan.Zero, maxWritesInFlight: maxWritesInFlight)); + driver, commsLossTimeout: TimeSpan.Zero, maxWritesInFlight: maxWritesInFlight, + queuedWriteBudget: queuedWriteBudget)); actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); parent.ExpectMsg(m => m.Connected, PresenceBudget); return (actor, parent); @@ -49,10 +51,10 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa results.ShouldAllBe(r => r.Success); driver.Writes.Select(w => w.FullReference).ShouldBe(new[] { "tag-1", "tag-2" }, ignoreOrder: true); - // Serialised writes take two latencies; concurrent ones take one. The margin covers a loaded machine. - sw.Elapsed.ShouldBeLessThan(WriteLatency + WriteLatency / 2, - "the second write waited for the first — writes are still serialised on the mailbox"); - driver.MaxConcurrentWrites.ShouldBe(2, "positive control: both writes were inside the driver at once"); + // Both were inside the driver at once — that is the proof; the wall clock is only a sanity ceiling + // (serialised writes would take two latencies, and the host's Ask is 8 s). + driver.MaxConcurrentWrites.ShouldBe(2, "the second write waited for the first — writes are still serialised on the mailbox"); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(8)); } [Fact] @@ -96,12 +98,69 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa driver.ReleaseWrites(); var results = await Task.WhenAll(asks); results.ShouldAllBe(r => r.Success); - // Entry order: the first two slots went to the first two writes (in either order — they are - // concurrent), the queued pair entered only after a slot freed, in queue order. + // Entry order: the first two slots went to the first two writes, the queued pair entered only after a + // slot freed. Within each pair the order is unspecified — distinct tags run concurrently on the pool; + // only SAME-tag order is guaranteed (pinned by the same-tag test below). var entered = driver.EnteredRefs; entered.Count.ShouldBe(4); entered.Take(2).ShouldBe(new[] { "tag-0", "tag-1" }, ignoreOrder: true); - entered.Skip(2).ShouldBe(new[] { "tag-2", "tag-3" }); + entered.Skip(2).ShouldBe(new[] { "tag-2", "tag-3" }, ignoreOrder: true); + } + + /// + /// Fix round 1: writes to the SAME tag must keep their order — two concurrent writes v1, v2 could land as + /// v1. A same-tag write waits behind its tag while other tags still run concurrently. + /// + [Fact] + public async Task Writes_to_the_same_tag_are_serialised_in_order_while_other_tags_run_concurrently() + { + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, _) = Connect(driver, maxWritesInFlight: 4); + + var first = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + var second = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-a", 2), PresenceBudget); + // Positive control: a DIFFERENT tag is not held behind tag-a. + var other = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-b", 9), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 2, PresenceBudget); + driver.EnteredRefs.ShouldBe(new[] { "tag-a", "tag-b" }); + + await Task.Delay(300); + driver.WritesEntered.ShouldBe(2, "the second tag-a write entered the driver while the first was still inside it"); + second.IsCompleted.ShouldBeFalse(); + + driver.ReleaseWrites(); + (await Task.WhenAll(first, second, other)).ShouldAllBe(r => r.Success); + driver.Writes.Where(w => w.FullReference == "tag-a").Select(w => w.Value).ShouldBe(new object[] { 1, 2 }); + } + + /// + /// Fix round 1: a write that sat in the queue longer than the host's Ask budget has already been reported + /// as a timeout (and reverted) upstream — writing it late would push a stale value to the device. + /// + [Fact] + public async Task A_queued_write_older_than_the_budget_is_failed_BadTimeout_instead_of_written_late() + { + var driver = new SlowWritableDriver { HoldWrites = true }; + var (actor, _) = Connect(driver, maxWritesInFlight: 1, queuedWriteBudget: TimeSpan.FromMilliseconds(300)); + + var inFlight = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-0", 0), PresenceBudget); + AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget); + var stale = actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-1", 1), PresenceBudget); + + await Task.Delay(600); + driver.ReleaseWrites(); + + (await inFlight).Success.ShouldBeTrue(); + var staleResult = await stale; + staleResult.Success.ShouldBeFalse(); + staleResult.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout"); + driver.EnteredRefs.ShouldBe(new[] { "tag-0" }, "the stale write must never reach the device"); } [Fact] 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 index 19f1b6e3..51914d87 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorReconnectSingleFlightTests.cs @@ -60,6 +60,32 @@ public sealed class DriverInstanceActorReconnectSingleFlightTests : RuntimeActor driver.Attempts.ShouldBe(2); } + /// + /// Fix round 1: the single-flight gate only cleared on the attempt's RESULT, and the invoker's timeout is + /// token-only — a driver whose InitializeAsync ignores its token wedged the reconnect loop forever. The + /// attempt is now bounded pessimistically (initializeTimeout); an overrun is abandoned, reported + /// as a failure, and the next tick retries. + /// + [Fact] + public void A_token_ignoring_never_completing_init_is_abandoned_and_a_later_attempt_connects() + { + var driver = new InterleavingDriver(initDuration: TimeSpan.Zero, failUntilAttempt: 1) + { + // Attempt 1 hangs forever and ignores cancellation; attempt 2 completes. + HangOnAttempt = 1, + }; + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, reconnectInterval: TimeSpan.FromMilliseconds(50), commsLossTimeout: TimeSpan.Zero, + initializeTimeout: TimeSpan.FromMilliseconds(300))); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + parent.FishForMessage(m => m is DriverInstanceActor.ConnectivityChanged { Connected: true }, PresenceBudget); + driver.Attempts.ShouldBe(2); + driver.HungAttemptStillRunning.ShouldBeTrue("positive control: the first attempt really was wedged"); + } + /// Records every Shutdown/Initialize entry and exit so an overlap is a fact, not an inference. private sealed class InterleavingDriver : IDriver { @@ -79,6 +105,13 @@ public sealed class DriverInstanceActorReconnectSingleFlightTests : RuntimeActor public Func? ShutdownBehavior { get; init; } + /// The 1-based attempt that never completes and ignores its token (0 = none). + public int HangOnAttempt { get; init; } + + private readonly TaskCompletionSource _never = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public bool HungAttemptStillRunning => !_never.Task.IsCompleted && Volatile.Read(ref _initsInProgress) > 0; + public int Attempts => Volatile.Read(ref _attempts); public int Overlaps => Volatile.Read(ref _overlaps); public string Interleaving { get { lock (_gate) return string.Join(" ", _events); } } @@ -93,6 +126,7 @@ public sealed class DriverInstanceActorReconnectSingleFlightTests : RuntimeActor Record($"init{attempt}>"); try { + if (attempt == HangOnAttempt) await _never.Task; if (_initDuration > TimeSpan.Zero) await Task.Delay(_initDuration, CancellationToken.None); if (attempt < _failUntilAttempt) throw new InvalidOperationException($"init attempt {attempt} refused"); }