fix(runtime,adminui): per-tag write ordering, pessimistic init bound, unique generations, PostStop budget (fix round 1)

- Writes: a second write to a tag already inside the driver waits in the queue behind it (_tagsInFlight) while other tags still run under the cap of 4; queued writes are stamped and failed BadTimeout at dequeue once older than the 8 s host Ask budget (Props queuedWriteBudget) instead of being written late. - Initialise attempts are bounded pessimistically (initializeTimeout, default 30 s = 2x Tier C) on top of the invoker's token-only timeout: an overrun is logged as abandoned, reported InitializeFailed so the retry loop continues, and its late result is dropped by generation. Generations come from a process-wide Interlocked counter so a prior incarnation's late result cannot collide (F6). - PostStop shares one shutdownTimeout budget across shutdown + dispose. - TelemetryDialSupervisor cancels the establishment deadline on StreamStopped; FleetDiagnosticsClient builds the ActorSelection inside the try (a malformed node degrades to "unavailable"); the diagnostics page labels the column "Last health publish". - Tests: same-tag serialisation with a distinct-tag positive control; stale queued write BadTimeout; token-ignoring hung init -> later attempt connects; stopwatch assertion demoted to a ceiling. Refs #731 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 10:36:30 -04:00
parent 59b72fe385
commit 953bd27533
6 changed files with 257 additions and 49 deletions
@@ -42,9 +42,11 @@ public sealed class FleetDiagnosticsClient : IFleetDiagnosticsClient
    /// <inheritdoc />
    public async Task<NodeDiagnosticsSnapshot> 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<NodeDiagnosticsSnapshot>(
@@ -120,7 +120,7 @@ else
                            <th>State</th>
                            <th class="numeric">Connected devices</th>
                            <th class="numeric">Faulted devices</th>
-                            <th>Last change</th>
+                            <th title="When the driver child last published a changed health snapshot (its spawn time until the first)">Last health publish</th>
                        </tr>
                    </thead>
                    <tbody>
@@ -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;
@@ -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
    /// </summary>
    public const int DefaultMaxWritesInFlight = 4;

-    /// <summary>Self-sent by a write task once it has replied, releasing its in-flight slot.</summary>
-    private sealed class WriteCompleted
-    {
-        public static readonly WriteCompleted Instance = new();
-        private WriteCompleted() { }
-    }
+    /// <summary>
+    ///     How long a queued write may wait for a slot before it is failed <c>BadTimeout</c> 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.
+    /// </summary>
+    public static readonly TimeSpan DefaultQueuedWriteBudget = TimeSpan.FromSeconds(8);
+
+    /// <summary>
+    ///     Pessimistic bound on one <see cref="IDriver.InitializeAsync"/> 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).
+    /// </summary>
+    public static readonly TimeSpan DefaultInitializeTimeout = TimeSpan.FromSeconds(30);
+
+    /// <summary>Self-sent by a write task once it has replied, releasing its slot and its tag.</summary>
+    private sealed record WriteCompleted(string TagId);

    private readonly IDriver _driver;

@@ -240,9 +251,20 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
    /// the self-sent <see cref="WriteCompleted"/>).</summary>
    private int _writesInFlight;

-    /// <summary>Writes waiting for an in-flight slot, in arrival order. Failed fast on every exit from
-    /// <c>Connected</c>, so a queued write never reaches a driver that is reconnecting.</summary>
-    private readonly Queue<(WriteAttribute Write, IActorRef ReplyTo)> _pendingWrites = new();
+    /// <summary>Writes waiting for a slot (or for their tag), in arrival order, stamped with arrival time.
+    /// Failed fast on every exit from <c>Connected</c>, so a queued write never reaches a driver that is
+    /// reconnecting; failed <c>BadTimeout</c> at dequeue once older than <see cref="_queuedWriteBudget"/>.</summary>
+    private readonly LinkedList<(WriteAttribute Write, IActorRef ReplyTo, DateTime EnqueuedUtc)> _pendingWrites = new();
+
+    /// <summary>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.</summary>
+    private readonly HashSet<string> _tagsInFlight = new(StringComparer.Ordinal);
+
+    /// <summary>See <see cref="DefaultQueuedWriteBudget"/>.</summary>
+    private readonly TimeSpan _queuedWriteBudget;
+
+    /// <summary>See <see cref="DefaultInitializeTimeout"/>.</summary>
+    private readonly TimeSpan _initializeTimeout;

    /// <summary>
    ///     Set by <see cref="PreRestart"/> so the <see cref="PostStop"/> 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.</summary>
    private int _initGeneration;

+    /// <summary>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.</summary>
+    private static int s_initGenerationSource;
+
    /// <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
@@ -376,6 +403,10 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
    /// hung shutdown is abandoned without waiting 5 s.</param>
    /// <param name="maxWritesInFlight">Optional cap on concurrent writes against the driver; defaults to
    /// <see cref="DefaultMaxWritesInFlight"/>. Writes beyond it queue in arrival order.</param>
+    /// <param name="queuedWriteBudget">Optional bound on how long a queued write may wait before it is failed
+    /// <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>
    /// <returns>A <see cref="Akka.Actor.Props"/> instance configured to create the wrapped <see cref="DriverInstanceActor"/>.</returns>
    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));

    /// <summary>
    /// 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
    /// <see cref="DefaultShutdownTimeout"/> when null.</param>
    /// <param name="maxWritesInFlight">Cap on concurrent writes; defaults to
    /// <see cref="DefaultMaxWritesInFlight"/> when null or non-positive.</param>
+    /// <param name="queuedWriteBudget">Bound on a queued write's wait; defaults to
+    /// <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>
    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<WriteAttribute>(_ =>
            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<WriteCompleted>(_ => OnWriteCompleted());
+        Receive<WriteCompleted>(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<RouteAlarmAck>(_ =>
@@ -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<WriteAttribute>(msg => StartOrQueueWrite(msg, Sender));
-        Receive<WriteCompleted>(_ => OnWriteCompleted());
+        Receive<WriteCompleted>(OnWriteCompleted);
        ReceiveAsync<RouteAlarmAck>(HandleAcknowledgeAsync);
        ReceiveAsync<Subscribe>(HandleSubscribeAsync);
        ReceiveAsync<Unsubscribe>(_ => UnsubscribeAsync());
@@ -846,7 +889,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
        // timeout on an inbound write to a transiently-down driver). Synchronous Receive.
        Receive<WriteAttribute>(_ =>
            Sender.Tell(new WriteAttributeResult(false, "driver not connected")));
-        Receive<WriteCompleted>(_ => OnWriteCompleted());
+        Receive<WriteCompleted>(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<RouteAlarmAck>(_ =>
@@ -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));
        }
    }

-    /// <summary>Frees a write slot and, while Connected, starts the next queued write.</summary>
-    private void OnWriteCompleted()
+    /// <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>
+    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.</summary>
    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<string, object?>("event", "stop"),
            new KeyValuePair<string, object?>("driver_type", _driver.DriverType));
    }

+    /// <summary>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.</summary>
+    private static readonly TimeSpan MinDisposeBudget = TimeSpan.FromMilliseconds(250);
+
    /// <summary>Final teardown of the wrapped driver: <see cref="IAsyncDisposable"/> preferred, else
-    /// <see cref="IDisposable"/>, never both. A driver implementing neither needs nothing here.</summary>
-    private void DisposeDriver()
+    /// <see cref="IDisposable"/>, 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
+    /// <see cref="IDriver.ShutdownAsync"/>.</summary>
+    /// <param name="bound">The remaining PostStop budget available to an async dispose.</param>
+    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)
        {
@@ -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<DriverInstanceActor.ConnectivityChanged>(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);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<DriverInstanceActor.WriteAttributeResult>(
+            new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget);
+        AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget);
+        var second = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
+            new DriverInstanceActor.WriteAttribute("tag-a", 2), PresenceBudget);
+        // Positive control: a DIFFERENT tag is not held behind tag-a.
+        var other = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
+            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 });
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<DriverInstanceActor.WriteAttributeResult>(
+            new DriverInstanceActor.WriteAttribute("tag-0", 0), PresenceBudget);
+        AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget);
+        var stale = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
+            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]
@@ -60,6 +60,32 @@ public sealed class DriverInstanceActorReconnectSingleFlightTests : RuntimeActor
        driver.Attempts.ShouldBe(2);
    }

+    /// <summary>
+    ///     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 (<c>initializeTimeout</c>); an overrun is abandoned, reported
+    ///     as a failure, and the next tick retries.
+    /// </summary>
+    [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");
+    }
+
    /// <summary>Records every Shutdown/Initialize entry and exit so an overlap is a fact, not an inference.</summary>
    private sealed class InterleavingDriver : IDriver
    {
@@ -79,6 +105,13 @@ public sealed class DriverInstanceActorReconnectSingleFlightTests : RuntimeActor

        public Func<CancellationToken, Task>? ShutdownBehavior { get; init; }

+        /// <summary>The 1-based attempt that never completes and ignores its token (0 = none).</summary>
+        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");
            }