perf(mqtt): publish a write batch concurrently, serialised per topic; writes feed health (R06:M-8)

WriteAsync published one awaited QoS 1 round-trip per request, so 50 writes to a
stalled broker cost 50 publish deadlines. Different topics now publish concurrently
(at most 8 in flight); same-topic writes stay in request order. A caller cancel still
fills every slot. A successful publish advances LastSuccessfulRead. MiniBroker now
frames packets properly and records/PUBACKs PUBLISHes.

Refs #744
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT

docs/drivers/Mqtt.md

@@ -491,6 +491,11 @@ with their real type; a birth type the encoder cannot represent (any `*Array`, D
 matters and is deliberately not configurable: a retained command would be replayed by the broker to the device on *every* future
 subscribe — including the device's own reconnect — so one stale write would be re-applied forever.
 
+**A batch publishes concurrently, one topic at a time (R06:M-8).** Writes to *different* topics in one
+`WriteAsync` go out together, at most 8 in flight, so a stalled broker costs a batch about one publish deadline
+rather than one per request; writes to the *same* topic stay sequential in request order, so the last one wins.
+A successful publish (PUBACK, or a sent QoS 0 command) advances `LastSuccessfulRead` like an inbound message.
+
 **Plain QoS 1 is at-least-once.** Know the consequences before pointing a Plain `writeTopic` at a non-idempotent
 device topic:
 

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs

@@ -104,6 +104,12 @@ public sealed class MqttDriver
     /// </summary>
     private const string DiscoveryFolderName = "Mqtt";
 
+    /// <summary>
+    ///     Ceiling on PUBLISHes one write batch keeps in flight (R06:M-8). Bounded so a large batch cannot
+    ///     flood a broker's inflight window; different topics share it, one topic's writes stay sequential.
+    /// </summary>
+    internal const int MaxConcurrentPublishes = 8;
+
     /// <summary>How often the host-connectivity poll samples <see cref="MqttConnection.State"/>.</summary>
     private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1);
 
@@ -209,6 +215,9 @@ public sealed class MqttDriver
 
     private DriverHealth _health = new(DriverState.Unknown, null, null);
 
+    /// <summary>UTC ticks of the last successful write PUBLISH; 0 = none. Answer evidence for health.</summary>
+    private long _lastPublishUtcTicks;
+
     /// <summary>Guards the generation bump against the generation-checked health write.</summary>
     private readonly object _healthLock = new();
 
@@ -446,7 +455,7 @@ public sealed class MqttDriver
             return stored;
         }
 
-        var lastMessage = connection.LastMessageUtc ?? stored.LastSuccessfulRead;
+        var lastMessage = Latest(connection.LastMessageUtc ?? stored.LastSuccessfulRead, LastPublishUtc);
         return connection.State switch
         {
             MqttConnectionState.Connected => new DriverHealth(DriverState.Healthy, lastMessage, stored.LastError),
@@ -708,9 +717,12 @@ public sealed class MqttDriver
     {
         ArgumentNullException.ThrowIfNull(writes);
 
-        var results = new WriteResult[writes.Count];
+        var results = new WriteResult?[writes.Count];
         var connection = _connection;
 
+        // Pass 1 — everything that needs no broker: resolve, encode, refuse. What survives is grouped by
+        // destination topic in REQUEST order.
+        var byTopic = new Dictionary<string, List<(int Index, MqttEncodedWrite Encoded, string Reference)>>(StringComparer.Ordinal);
         for (var i = 0; i < writes.Count; i++)
         {
             var w = writes[i];
@@ -738,22 +750,70 @@ public sealed class MqttDriver
                 continue;
             }
 
+            if (!byTopic.TryGetValue(encoded.Topic, out var queue))
+            {
+                byTopic[encoded.Topic] = queue = [];
+            }
+
+            queue.Add((i, encoded, w.FullReference));
+        }
+
+        // Pass 2 — publish (R06:M-8). Different topics go out concurrently, at most MaxConcurrentPublishes
+        // in flight, so a stalled broker costs a batch ~one publish deadline instead of one per request.
+        // Writes to the SAME topic stay strictly sequential in request order: the device applies them in
+        // arrival order, and the last one must win.
+        if (byTopic.Count > 0)
+        {
+            using var slots = new SemaphoreSlim(MaxConcurrentPublishes, MaxConcurrentPublishes);
+            await Task.WhenAll(byTopic.Values.Select(queue => PublishQueueAsync(connection!, queue, slots, results, cancellationToken)))
+                .ConfigureAwait(false);
+        }
+
+        // Every slot is filled — a caller cancellation leaves the unattempted ones BadCommunicationError
+        // rather than throwing, so the batch shape holds.
+        var filled = new WriteResult[writes.Count];
+        for (var i = 0; i < filled.Length; i++)
+        {
+            filled[i] = results[i] ?? new WriteResult(StatusBadCommunicationError);
+        }
+
+        return filled;
+    }
+
+    /// <summary>Publishes one topic's writes in request order, each under a shared concurrency slot.</summary>
+    private async Task PublishQueueAsync(
+        MqttConnection connection,
+        List<(int Index, MqttEncodedWrite Encoded, string Reference)> queue,
+        SemaphoreSlim slots,
+        WriteResult?[] results,
+        CancellationToken cancellationToken)
+    {
+        foreach (var (index, encoded, reference) in queue)
+        {
+            try
+            {
+                await slots.WaitAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
+                return;   // caller-cancelled: the remaining slots are filled by the caller
+            }
+
             try
             {
                 await connection
                     .PublishAsync(encoded.Topic, encoded.Payload, encoded.Qos, retain: false, cancellationToken)
                     .ConfigureAwait(false);
-                results[i] = new WriteResult(0u);
+                results[index] = new WriteResult(0u);
+
+                // A PUBACK (or a sent QoS 0 command) is the broker answering — evidence the session works,
+                // so it feeds LastSuccessfulRead alongside inbound messages.
+                Interlocked.Exchange(ref _lastPublishUtcTicks, DateTime.UtcNow.Ticks);
             }
             catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
             {
-                // Caller-cancelled: fill the remainder rather than throwing, so the batch shape holds.
-                for (var j = i; j < writes.Count; j++)
-                {
-                    results[j] = new WriteResult(StatusBadCommunicationError);
-                }
-
-                return results;
+                results[index] = new WriteResult(StatusBadCommunicationError);
+                return;
             }
             catch (Exception ex)
             {
@@ -761,13 +821,15 @@ public sealed class MqttDriver
                     ex,
                     "MQTT driver '{DriverId}': write to '{Reference}' could not be published on '{Topic}'.",
                     _driverInstanceId,
-                    w.FullReference,
+                    reference,
                     encoded.Topic);
-                results[i] = new WriteResult(StatusBadCommunicationError);
+                results[index] = new WriteResult(StatusBadCommunicationError);
+            }
+            finally
+            {
+                slots.Release();
             }
         }
-
-        return results;
     }
 
     /// <summary>Maps an encode refusal to the status an OPC UA client sees.</summary>
@@ -1528,6 +1590,20 @@ public sealed class MqttDriver
         OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, previous, next));
     }
 
+    /// <summary>The last successful write publish, or null.</summary>
+    private DateTime? LastPublishUtc
+    {
+        get
+        {
+            var ticks = Interlocked.Read(ref _lastPublishUtcTicks);
+            return ticks == 0 ? null : new DateTime(ticks, DateTimeKind.Utc);
+        }
+    }
+
+    /// <summary>The later of two optional instants.</summary>
+    private static DateTime? Latest(DateTime? a, DateTime? b)
+        => a is null ? b : b is null ? a : (a > b ? a : b);
+
     /// <summary>Barrier-protected read of the multi-threaded health field.</summary>
     private DriverHealth ReadHealth() => Volatile.Read(ref _health);
 

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MiniBroker.cs

@@ -5,7 +5,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
 
 /// <summary>
 ///     The smallest broker that can complete an MQTT 3.1.1 handshake: accept TCP, answer CONNECT with
-///     a CONNACK, answer PINGREQ with PINGRESP, and nothing else. That is enough for MQTTnet to report
+///     a CONNACK, answer PINGREQ with PINGRESP, SUBSCRIBE with a SUBACK (unless told not to), and record
+///     PUBLISHes (PUBACKing QoS 1 only when told to). It never forwards a PUBLISH to a subscriber. That is enough for MQTTnet to report
 ///     a real connection — which is what makes a real drop (<see cref="DropAll"/>) drive a real
 ///     <c>DisconnectedAsync</c> and a real reconnect, rather than a test seam standing in for one.
 /// </summary>
@@ -107,6 +108,96 @@ internal sealed class MiniBroker : IDisposable
         return [.. ack];
     }
 
+    /// <summary>
+    ///     When true, a QoS 1 PUBLISH is answered with a PUBACK. False (the default) is a broker that takes
+    ///     the PUBLISH and never acknowledges it — the stalled-broker shape a write deadline exists for.
+    /// </summary>
+    public bool AckPublish { get; set; }
+
+    /// <summary>Every PUBLISH received, in arrival order, as <c>(topic, payload)</c>.</summary>
+    public System.Collections.Concurrent.ConcurrentQueue<(string Topic, byte[] Payload)> Published { get; } = new();
+
+    /// <summary>Handles one complete packet; returns false when the connection must close.</summary>
+    private async Task<bool> HandlePacketAsync(NetworkStream stream, byte[] packet)
+    {
+        var packetType = packet[0] & 0xF0;
+        if (packetType == 0x10)
+        {
+            await stream.WriteAsync(new byte[] { 0x20, 0x02, 0x00, ConnAckReturnCode }, _cts.Token);
+
+            // A real broker closes the socket straight after refusing, which is what drives
+            // MQTTnet's DisconnectedAsync. Mirroring it keeps the fake honest.
+            return ConnAckReturnCode == ReturnCodeAccepted;
+        }
+
+        if (packetType == 0xC0)
+        {
+            await stream.WriteAsync(PingResp, _cts.Token);
+        }
+        else if (packetType == 0x80 && AnswerSubscribe && TryBuildSubAck(packet, packet.Length) is { } subAck)
+        {
+            await stream.WriteAsync(subAck, _cts.Token);
+        }
+        else if (packetType == 0x30)
+        {
+            var qos = (packet[0] >> 1) & 0x03;
+            var at = HeaderLength(packet);
+            var topicLength = (packet[at] << 8) | packet[at + 1];
+            var topic = System.Text.Encoding.UTF8.GetString(packet, at + 2, topicLength);
+            at += 2 + topicLength;
+            byte idHigh = 0, idLow = 0;
+            if (qos > 0)
+            {
+                idHigh = packet[at];
+                idLow = packet[at + 1];
+                at += 2;
+            }
+
+            Published.Enqueue((topic, packet[at..]));
+            if (qos == 1 && AckPublish)
+            {
+                await stream.WriteAsync(new byte[] { 0x40, 0x02, idHigh, idLow }, _cts.Token);
+            }
+        }
+
+        return true;
+    }
+
+    /// <summary>Length of a packet's fixed header (type byte + variable-length remaining length).</summary>
+    private static int HeaderLength(byte[] packet)
+    {
+        var at = 1;
+        while ((packet[at] & 0x80) != 0) at++;
+        return at + 1;
+    }
+
+    /// <summary>Removes one complete packet from the front of <paramref name="pending"/>, if one is there.</summary>
+    private static bool TryTakePacket(List<byte> pending, out byte[] packet)
+    {
+        packet = [];
+        if (pending.Count < 2) return false;
+
+        var at = 1;
+        var remaining = 0;
+        var multiplier = 1;
+        byte digit;
+        do
+        {
+            if (at >= pending.Count) return false;
+            digit = pending[at++];
+            remaining += (digit & 0x7F) * multiplier;
+            multiplier *= 128;
+        }
+        while ((digit & 0x80) != 0);
+
+        var total = at + remaining;
+        if (pending.Count < total) return false;
+
+        packet = pending.GetRange(0, total).ToArray();
+        pending.RemoveRange(0, total);
+        return true;
+    }
+
     /// <summary>Total TCP connections accepted since construction — counts reconnect attempts.</summary>
     public int AcceptedCount => Volatile.Read(ref _acceptedCount);
 
@@ -178,7 +269,8 @@ internal sealed class MiniBroker : IDisposable
 
     private async Task ServeAsync(TcpClient client)
     {
-        var buffer = new byte[1024];
+        var buffer = new byte[4096];
+        var pending = new List<byte>();
         try
         {
             var stream = client.GetStream();
@@ -195,25 +287,22 @@ internal sealed class MiniBroker : IDisposable
                     continue;
                 }
 
-                var packetType = buffer[0] & 0xF0;
-                if (packetType == 0x10)
+                // Frame properly: several packets can share one TCP read (concurrent publishes do), and
+                // one packet can span two.
+                pending.AddRange(buffer.AsSpan(0, read).ToArray());
+                var close = false;
+                while (TryTakePacket(pending, out var packet))
                 {
-                    await stream.WriteAsync(new byte[] { 0x20, 0x02, 0x00, ConnAckReturnCode }, _cts.Token);
-
-                    // A real broker closes the socket straight after refusing, which is what drives
-                    // MQTTnet's DisconnectedAsync. Mirroring it keeps the fake honest.
-                    if (ConnAckReturnCode != ReturnCodeAccepted)
+                    if (!await HandlePacketAsync(stream, packet))
                     {
+                        close = true;
                         break;
                     }
                 }
-                else if (packetType == 0xC0)
+
+                if (close)
                 {
-                    await stream.WriteAsync(PingResp, _cts.Token);
-                }
-                else if (packetType == 0x80 && AnswerSubscribe && TryBuildSubAck(buffer, read) is { } subAck)
-                {
-                    await stream.WriteAsync(subAck, _cts.Token);
+                    break;
                 }
             }
         }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs

@@ -288,6 +288,79 @@ public sealed class MqttDriverSessionLifecycleTests
         driver.GetHealth().State.ShouldBe(DriverState.Healthy);
     }
 
+    private static RawTagEntry WritableTag(string rawPath, string writeTopic) => new(
+        rawPath,
+        $$"""{"topic":"{{writeTopic}}/state","payloadFormat":"Scalar","dataType":"Float64","writeTopic":"{{writeTopic}}"}""",
+        WriteIdempotent: false);
+
+    /// <summary>
+    ///     <b>R06:M-8.</b> A write batch used to publish strictly serially, one awaited QoS 1 round-trip per
+    ///     request — against a stalled broker, N writes × the publish deadline. Different topics now publish
+    ///     concurrently: all four PUBLISHes reach a broker that never PUBACKs well inside ONE 30 s deadline
+    ///     (serial code would have sent exactly one). Cancelling the batch then fills every slot, so the
+    ///     batch shape holds.
+    /// </summary>
+    [Fact]
+    public async Task AWriteBatch_PublishesDifferentTopicsConcurrently()
+    {
+        using var broker = new MiniBroker();   // never PUBACKs
+        var options = BrokerOptions(broker.Port) with
+        {
+            ConnectTimeoutSeconds = 30,
+            RawTags = [.. Enumerable.Range(0, 4).Select(i => WritableTag($"Plant/Mqtt/dev1/Sp{i}", $"cmd/sp{i}"))],
+        };
+        await using var driver = new MqttDriver(options, "d", null);
+        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
+
+        using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
+        var write = driver.WriteAsync(
+            [.. Enumerable.Range(0, 4).Select(i => new WriteRequest($"Plant/Mqtt/dev1/Sp{i}", (double)i))],
+            cts.Token);
+
+        await WaitForAsync(() => broker.Published.Count == 4, "all four PUBLISHes to be in flight at once");
+        write.IsCompleted.ShouldBeFalse("no PUBACK has arrived, so nothing can have completed");
+
+        await cts.CancelAsync();
+        var results = await write.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
+        results.Count.ShouldBe(4);
+        results.ShouldAllBe(r => r.StatusCode == 0x80050000u);   // BadCommunicationError, every slot filled
+    }
+
+    /// <summary>
+    ///     Writes to the SAME topic in one batch stay in request order — last write wins at the device —
+    ///     while the batch as a whole is concurrent. The broker PUBACKs here, so every slot is Good — and
+    ///     those acks advance <c>LastSuccessfulRead</c> (writes feed health).
+    /// </summary>
+    [Fact]
+    public async Task AWriteBatch_KeepsSameTopicWritesInRequestOrder()
+    {
+        using var broker = new MiniBroker { AckPublish = true };
+        var options = BrokerOptions(broker.Port) with
+        {
+            RawTags = [WritableTag("Plant/Mqtt/dev1/Sp", "cmd/sp"), WritableTag("Plant/Mqtt/dev1/Other", "cmd/other")],
+        };
+        await using var driver = new MqttDriver(options, "d", null);
+        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
+        driver.GetHealth().LastSuccessfulRead.ShouldBeNull("nothing has answered yet");
+
+        var results = await driver.WriteAsync(
+            [
+                new WriteRequest("Plant/Mqtt/dev1/Sp", 1.0),
+                new WriteRequest("Plant/Mqtt/dev1/Other", 9.0),
+                new WriteRequest("Plant/Mqtt/dev1/Sp", 2.0),
+                new WriteRequest("Plant/Mqtt/dev1/Sp", 3.0),
+            ],
+            TestContext.Current.CancellationToken);
+
+        results.ShouldAllBe(r => r.StatusCode == 0u);
+        broker.Published.Where(p => p.Topic == "cmd/sp")
+            .Select(p => System.Text.Encoding.UTF8.GetString(p.Payload))
+            .ShouldBe(["1", "2", "3"]);
+
+        // A PUBACK is the broker answering, so writes feed the evidence clock the comms-loss watchdog reads.
+        driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
+    }
+
     private static async Task WaitForAsync(Func<bool> condition, string what)
     {
         var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);