Merge review R5 fixes: MTConnect session publish + generation bump atomic, superseded pumps joined on stop; MQTT writes encode the birth's own Sparkplug datatype (unsupported birth types refused BadNotSupported); AbLegacy dead init join removed + Dispose join test; MQTT QoS1 and Postgres key-normalisation documented

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs

@@ -239,14 +239,10 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
             _health = new DriverHealth(DriverState.Faulted, null, ex.Message);
             _logger.LogError(ex,
                 "AbLegacy driver initialise failed. Driver={DriverInstanceId}", _driverInstanceId);
-            // Tear down any probe loops and cached state that were created before the failure so
-            // that a caller who catches and abandons (rather than retrying via ReinitializeAsync)
-            // doesn't leave orphaned background tasks, CancellationTokenSources, and libplctag
-            // handles alive. Mirrors the body of ShutdownAsync without awaiting the poll engine
-            // (nothing has been subscribed yet at init time).
-            foreach (var state in _devices.Values)
-                try { state.ProbeCts?.Cancel(); } catch { }
-            JoinProbeLoopsSync();
+            // Release cached state created before the failure so a caller who catches and abandons
+            // (rather than retrying via ReinitializeAsync) doesn't leave CancellationTokenSources and
+            // libplctag handles alive. No probe-loop join here: the loops start as the LAST statement of
+            // the try and nothing after that can throw, so a failure never has a running loop to join.
             foreach (var state in _devices.Values)
             {
                 state.ProbeCts?.Dispose();

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs

@@ -341,6 +341,14 @@ public sealed class MTConnectDriver
     /// </summary>
     private long _sessionGeneration;
 
+    /// <summary>
+    ///     Pumps retired by <see cref="StartSampleStreamCore"/> because they were bound to a replaced
+    ///     session (STAB-20), kept so <see cref="StopSampleStreamAsync"/> can join them — a pump nobody
+    ///     awaits can still run its default-verdict arm (<c>Degrade</c>, the reconnect counter) after
+    ///     the new session has written Healthy. Guarded by <see cref="_subscriptionLock"/>.
+    /// </summary>
+    private readonly List<(CancellationTokenSource Cts, Task Task)> _supersededPumps = [];
+
     /// <summary>
     ///     The Agent answered <c>/sample</c> with something that cannot be streamed at all, so the
     ///     pump gave up. Latched — reconnecting reproduces the identical answer forever, and a
@@ -1806,6 +1814,33 @@ public sealed class MTConnectDriver
         }
     }
 
+    /// <summary>Disposes and forgets superseded pumps that have finished. <b>Caller must hold <see cref="_subscriptionLock"/>.</b></summary>
+    private void PruneSupersededPumpsCore()
+    {
+        for (var i = _supersededPumps.Count - 1; i >= 0; i--)
+        {
+            if (!_supersededPumps[i].Task.IsCompleted)
+            {
+                continue;
+            }
+
+            _supersededPumps[i].Cts.Dispose();
+            _supersededPumps.RemoveAt(i);
+        }
+    }
+
+    /// <summary>The retired-but-unjoined pump tasks (test barrier).</summary>
+    internal Task[] SupersededPumpTasks
+    {
+        get
+        {
+            lock (_subscriptionLock)
+            {
+                return [.. _supersededPumps.Select(p => p.Task)];
+            }
+        }
+    }
+
     /// <summary>
     ///     Starts the shared <c>/sample</c> pump, if there is anything to pump and anything to pump
     ///     from. <b>Caller must hold <see cref="_subscriptionLock"/>.</b>
@@ -1836,6 +1871,16 @@ public sealed class MTConnectDriver
             try { _pumpCts?.Cancel(); }
             catch (ObjectDisposedException) { /* already going away */ }
 
+            // Retain (don't drop) the retired pump so shutdown can join it; its source is disposed
+            // by whoever observes it complete.
+            PruneSupersededPumpsCore();
+            if (_pumpCts is not null && _pumpTask is not null)
+            {
+                _supersededPumps.Add((_pumpCts, _pumpTask));
+                _pumpCts = null;
+            }
+            _pumpTask = null;
+
             _logger.LogInformation(
                 "MTConnect driver {DriverInstanceId} replaced a /sample pump bound to a retired Agent session.",
                 _driverInstanceId);
@@ -2726,15 +2771,17 @@ public sealed class MTConnectDriver
 
             // One publish, so a concurrent lock-free reader sees the new client, the new probe
             // cache, and the pump's opening cursor together or none of them.
+            // The generation bump and the session write are ONE critical section: a Subscribe landing
+            // between them would start a pump on the OLD session stamped with the NEW generation, and
+            // the restart's own StartSampleStreamCore would then treat it as current.
             lock (_subscriptionLock)
             {
+                Volatile.Write(
+                    ref _session,
+                    new AgentSession(client, probe, current.InstanceId, current.NextSequence));
                 _sessionGeneration++;
             }
 
-            Volatile.Write(
-                ref _session,
-                new AgentSession(client, probe, current.InstanceId, current.NextSequence));
-
             var index = new MTConnectObservationIndex(binding.IndexTags);
             index.Apply(current);
             Volatile.Write(ref _index, index);
@@ -2864,6 +2911,7 @@ public sealed class MTConnectDriver
     {
         CancellationTokenSource? cts;
         Task? task;
+        (CancellationTokenSource Cts, Task Task)[] retired;
 
         lock (_subscriptionLock)
         {
@@ -2871,6 +2919,27 @@ public sealed class MTConnectDriver
             task = _pumpTask;
             _pumpCts = null;
             _pumpTask = null;
+            retired = [.. _supersededPumps];
+            _supersededPumps.Clear();
+        }
+
+        // Join pumps retired by a session replacement (already cancelled): they must not outlive
+        // teardown and write health / the reconnect counter over the next session.
+        foreach (var (retiredCts, retiredTask) in retired)
+        {
+            try
+            {
+                await retiredCts.CancelAsync().ConfigureAwait(false);
+                await retiredTask.WaitAsync(TeardownWait, cancellationToken).ConfigureAwait(false);
+                retiredCts.Dispose();
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(
+                    ex,
+                    "MTConnect driver {DriverInstanceId} could not join a superseded /sample pump within {TeardownWait}; it is cancelled and abandoned.",
+                    _driverInstanceId, TeardownWait);
+            }
         }
 
         if (cts is null)

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs

@@ -102,6 +102,14 @@ public static class MqttWriteEncoder
     ///     <see cref="MqttTagDefinition.DataType"/>, which for an unauthored tag is only the record's
     ///     default and would command the metric with the wrong wire type. Ignored in Plain mode.
     /// </param>
+    /// <param name="birthDataType">
+    ///     Sparkplug only: the metric's OWN birth-declared <see cref="SparkplugDataType"/>, used when the tag
+    ///     authored no type. Preferred over <paramref name="effectiveDataType"/> because the
+    ///     <c>ToDriverDataType</c> map is lossy (Int8→Int16, Uint8→Uint16, Text/Uuid/Bytes/File→String,
+    ///     <c>*Array</c>→scalar), so commanding through it sends the wrong Tahu datatype. A birth type the
+    ///     encoder cannot represent (arrays, DataSet, Template, …) is REFUSED as
+    ///     <see cref="MqttWriteEncodeError.UnsupportedDataType"/> — a wrong type is never sent.
+    /// </param>
     /// <returns>The encode outcome.</returns>
     /// <remarks>
     ///     <b>Retain <see langword="false"/> always; QoS 1 for Plain, 0 for Sparkplug commands.</b> Retain is the one that matters: a retained
@@ -115,13 +123,14 @@ public static class MqttWriteEncoder
         MqttTagDefinition def,
         object? value,
         out MqttEncodedWrite encoded,
-        DriverDataType? effectiveDataType = null)
+        DriverDataType? effectiveDataType = null,
+        SparkplugDataType? birthDataType = null)
     {
         ArgumentNullException.ThrowIfNull(def);
         encoded = default;
 
         return mode == MqttMode.SparkplugB
-            ? TryEncodeSparkplug(def, value, effectiveDataType, out encoded)
+            ? TryEncodeSparkplug(def, value, effectiveDataType, birthDataType, out encoded)
             : TryEncodePlain(def, value, out encoded);
     }
 
@@ -195,6 +204,7 @@ public static class MqttWriteEncoder
         MqttTagDefinition def,
         object? value,
         DriverDataType? effectiveDataType,
+        SparkplugDataType? birthDataType,
         out MqttEncodedWrite encoded)
     {
         encoded = default;
@@ -226,7 +236,9 @@ public static class MqttWriteEncoder
         }
 
         var metric = new Payload.Types.Metric { Name = metricName };
-        var error = ApplyValue(metric, def.DataTypeAuthored ? def.DataType : effectiveDataType ?? def.DataType, value);
+        var error = !def.DataTypeAuthored && birthDataType is { } birth
+            ? ApplyBirthValue(metric, birth, value)
+            : ApplyValue(metric, def.DataTypeAuthored ? def.DataType : effectiveDataType ?? def.DataType, value);
         if (error != MqttWriteEncodeError.None)
         {
             return error;
@@ -251,6 +263,63 @@ public static class MqttWriteEncoder
         return MqttWriteEncodeError.None;
     }
 
+    /// <summary>
+    ///     Sets the metric's value field and <c>Datatype</c> from the metric's own birth-declared Sparkplug
+    ///     type, for the types whose driver-type mapping is lossy. Types with a 1:1 driver mapping delegate
+    ///     to <see cref="ApplyValue"/>.
+    /// </summary>
+    /// <param name="metric">The metric being built.</param>
+    /// <param name="birth">The birth certificate's declared datatype.</param>
+    /// <param name="value">The value to convert.</param>
+    /// <returns>The encode outcome.</returns>
+    private static MqttWriteEncodeError ApplyBirthValue(Payload.Types.Metric metric, SparkplugDataType birth, object value)
+    {
+        try
+        {
+            switch (birth)
+            {
+                case TahuDataType.Int8:
+                    metric.Datatype = (uint)TahuDataType.Int8;
+                    metric.IntValue = unchecked((uint)Convert.ToSByte(value, CultureInfo.InvariantCulture));
+                    return MqttWriteEncodeError.None;
+
+                case TahuDataType.Uint8:
+                    metric.Datatype = (uint)TahuDataType.Uint8;
+                    metric.IntValue = Convert.ToByte(value, CultureInfo.InvariantCulture);
+                    return MqttWriteEncodeError.None;
+
+                case TahuDataType.Text:
+                case TahuDataType.Uuid:
+                    metric.Datatype = (uint)birth;
+                    metric.StringValue = value as string ?? Convert.ToString(value, CultureInfo.InvariantCulture) ?? "";
+                    return MqttWriteEncodeError.None;
+
+                case TahuDataType.Bytes:
+                case TahuDataType.File:
+                    if (value is not byte[] bytes)
+                    {
+                        return MqttWriteEncodeError.ValueNotRepresentable;
+                    }
+
+                    metric.Datatype = (uint)birth;
+                    metric.BytesValue = ByteString.CopyFrom(bytes);
+                    return MqttWriteEncodeError.None;
+            }
+        }
+        catch (Exception ex) when (ex is InvalidCastException or FormatException or OverflowException)
+        {
+            return MqttWriteEncodeError.ValueNotRepresentable;
+        }
+
+        // Arrays, DataSet, Template, PropertySet, Unknown: refused — never sent as a scalar of another type.
+        if (birth.IsSparkplugArray() || birth.ToDriverDataType() is null)
+        {
+            return MqttWriteEncodeError.UnsupportedDataType;
+        }
+
+        return ApplyValue(metric, birth.ToDriverDataType()!.Value, value);
+    }
+
     /// <summary>
     ///     Sets the metric's typed value field and its <c>Datatype</c> from the tag's declared type.
     /// </summary>

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

@@ -88,6 +88,9 @@ public sealed class MqttDriver
     /// </summary>
     private const uint StatusBadNotWritable = 0x803B0000u;
 
+    /// <summary>OPC UA <c>BadNotSupported</c> — the value's wire type has no safe command encoding.</summary>
+    private const uint StatusBadNotSupported = 0x803D0000u;
+
     /// <summary>The supplied value cannot be represented in the tag's declared data type.</summary>
     private const uint StatusBadTypeMismatch = 0x80740000u;
 
@@ -579,6 +582,24 @@ public sealed class MqttDriver
         return table?.ResolveByName(metric)?.DataType.ToDriverDataType() ?? def.DataType;
     }
 
+    /// <summary>
+    ///     The metric's own birth-declared Sparkplug datatype (unmapped), or <see langword="null"/> when the
+    ///     tag authored a type, is not a complete Sparkplug binding, or no birth has been seen. Fed to the
+    ///     write encoder so lossy driver-type mappings (Int8, Text, Bytes, arrays, …) are not commanded with
+    ///     the wrong wire type.
+    /// </summary>
+    private Org.Eclipse.Tahu.Protobuf.DataType? ResolveBirthDataType(MqttTagDefinition def)
+    {
+        if (def.DataTypeAuthored || _sparkplug is not { } sparkplug
+            || def.GroupId is not { } groupId || def.EdgeNodeId is not { } edgeNodeId || def.MetricName is not { } metric)
+        {
+            return null;
+        }
+
+        return sparkplug.Births.Find(new SparkplugScope(groupId, edgeNodeId, def.DeviceId))
+            ?.ResolveByName(metric)?.DataType;
+    }
+
     // ---- ISubscribable (straight through to the manager) ----
 
     /// <inheritdoc />
@@ -718,7 +739,7 @@ public sealed class MqttDriver
             }
 
             var error = MqttWriteEncoder.TryEncode(
-                _options.Mode, def, w.Value, out var encoded, ResolveDiscoveredDataType(def));
+                _options.Mode, def, w.Value, out var encoded, ResolveDiscoveredDataType(def), ResolveBirthDataType(def));
             if (error != MqttWriteEncodeError.None)
             {
                 results[i] = new WriteResult(MapEncodeError(error));
@@ -772,7 +793,7 @@ public sealed class MqttDriver
     private static uint MapEncodeError(MqttWriteEncodeError error) => error switch
     {
         MqttWriteEncodeError.NoWriteDestination => StatusBadNotWritable,
-        MqttWriteEncodeError.UnsupportedDataType => StatusBadNotWritable,
+        MqttWriteEncodeError.UnsupportedDataType => StatusBadNotSupported,
         _ => StatusBadTypeMismatch,
     };