fix(drivers): R5 minors - MTConnect session publish+generation atomic and superseded pump joined; MQTT birth-typed writes use the birth's own datatype; AbLegacy dead init join removed + Dispose join test; docs (#602)

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

docs/drivers/Mqtt.md

@@ -467,10 +467,27 @@ Both publish with `retain: false`. A **Plain** write goes at **QoS 1** (STAB-26)
 once the broker PUBACKs it, so `Good` means "the broker accepted it", and a broker refusal or a connection
 that died under the send is `Bad`. **Sparkplug commands stay at QoS 0** — a redelivered NCMD/DCMD would be
 re-applied. A Sparkplug metric the tag did not give a `dataType` is commanded with the type its live birth
-certificate declared (STAB-25), the same authored-beats-birth precedence reads use. Retain is the one that
+certificate declared (STAB-25), the same authored-beats-birth precedence reads use — using the birth's **own**
+Sparkplug datatype, not the lossy driver-type mapping, so Int8/UInt8/Text/Uuid/Bytes/File metrics are commanded
+with their real type; a birth type the encoder cannot represent (any `*Array`, DataSet, Template, …) is refused
+`BadNotSupported` rather than sent as another type. Retain is the one that
 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.
 
+**Plain QoS 1 is at-least-once.** Know the consequences before pointing a Plain `writeTopic` at a non-idempotent
+device topic:
+
+- A lost PUBACK can make a device subscribed at QoS ≥ 1 with a persistent session receive the command **twice**
+  (at QoS 0 delivery was downgraded and a duplicate could not happen).
+- A lost PUBACK, or the `ConnectTimeoutSeconds` (15 s) deadline, reports `Bad` and the node is **reverted**, even
+  though the broker may already have delivered the write. `Good` is "the broker accepted it", `Bad` is "not
+  confirmed", never "not delivered".
+- Each write holds `DriverInstanceActor`'s mailbox until the PUBACK arrives — up to 15 s per write, one after
+  another; comparable to a Modbus round trip.
+- A broker advertising MQTT 5 Maximum QoS 0 will disconnect on every Plain write.
+
+There is no per-tag QoS option; make the receiving topic idempotent (absolute setpoints, not increments).
+
 ### `writeTopic` is separate from `topic`, and its absence is meaningful
 
 A tag commonly reads `…/state` and writes `…/set`, so the destination is its own authored field. A

docs/drivers/Sql.md

@@ -95,6 +95,13 @@ that holds an INTEGER station id can be authored either `"7"` or `7`. A JSON obj
 is not a value and is refused. On **PostgreSQL** a string bind is sent untyped (`NpgsqlDbType.Unknown`,
 `PostgresDialect.ConfigureParameter`, STAB-24) so the server infers its type from the compared column;
 bound as `text`, a `"7"` against an `integer` key fails with `operator does not exist: integer = text`.
+
+⚠️ **Known limitation — key normalisation.** Row slice-back matches the returned key cell by its *text* form
+(`Convert.ToString`, ordinal), while the server now matches by the column's inferred type. An authored key the
+server treats as equal but whose text differs therefore selects a row that the driver then cannot find, and the
+tag reads **no-row**: `"007"` against an `integer` column (returns `7`), `"1.50"` against `numeric` (returns
+`1.5`), and an upper-case GUID against a `uuid` column (returned lower-case). Author keys in the column's
+canonical text form. SQL Server behaves the same way.
 (They were read asymmetrically until 2026-08-10, so a numerically
 authored `keyValue` refused the tag while the identically-bound `whereValue` was served — and the
 deploy gate reported the key as *missing* when it was right there in the blob.)

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
+            // Release cached state created before the failure so a caller who catches and abandons
-            // that a caller who catches and abandons (rather than retrying via ReinitializeAsync)
+            // (rather than retrying via ReinitializeAsync) doesn't leave CancellationTokenSources and
-            // doesn't leave orphaned background tasks, CancellationTokenSources, and libplctag
+            // libplctag handles alive. No probe-loop join here: the loops start as the LAST statement of
-            // handles alive. Mirrors the body of ShutdownAsync without awaiting the poll engine
+            // the try and nothing after that can throw, so a failure never has a running loop to join.
-            // (nothing has been subscribed yet at init time).
-            foreach (var state in _devices.Values)
-                try { state.ProbeCts?.Cancel(); } catch { }
-            JoinProbeLoopsSync();
             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,
     };
 

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyShutdownJoinTests.cs

@@ -51,4 +51,34 @@ public sealed class AbLegacyShutdownJoinTests
 
         factory.Tags["S:0"].Disposed.ShouldBeTrue();
     }
+
+    /// <summary>The synchronous <c>Dispose()</c> path joins the parked loop too (it used to cancel and dispose under it).</summary>
+    [Fact]
+    public async Task Dispose_WaitsForProbeLoopParkedInInit_ThenProbeRuntimeIsDisposed()
+    {
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var factory = new FakeAbLegacyTagFactory
+        {
+            Customise = p => new ParkedInitTag(p, entered, release),
+        };
+        var drv = new AbLegacyDriver(new AbLegacyDriverOptions
+        {
+            Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0", AbLegacyPlcFamily.Slc500)],
+            Probe = new AbLegacyProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(50), ProbeAddress = "S:0" },
+        }, "drv-join-dispose", factory);
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        var dispose = Task.Run(drv.Dispose);
+
+        await Task.Delay(300);
+        dispose.IsCompleted.ShouldBeFalse("Dispose must join the probe loop parked in runtime init");
+
+        // Positive control: releasing the parked call lets Dispose finish, with the probe runtime released.
+        release.SetResult();
+        await dispose.WaitAsync(TimeSpan.FromSeconds(5));
+
+        factory.Tags["S:0"].Disposed.ShouldBeTrue();
+    }
 }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs

@@ -35,6 +35,7 @@ public sealed class MTConnectPumpGenerationTests
             "mt-gen",
             _ => clients.Dequeue());
         await driver.InitializeAsync("{}", Ct);
+        Task[] retired = [];
 
         var blockOnce = 0;
         var blocking = new ManualResetEventSlim(false);
@@ -75,6 +76,11 @@ public sealed class MTConnectPumpGenerationTests
             blocking.Set();
             await restart.WaitAsync(Watchdog, Ct);
 
+            // The pump retired by the restart is retained so shutdown can join it (it used to be dropped
+            // un-awaited and could write health after the new session was Healthy).
+            retired = driver.SupersededPumpTasks;
+            retired.ShouldNotBeEmpty();
+
             // The restart installed a new client; the standing subscription must be streaming from IT.
             await clientB.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
             clientA.IsDisposed.ShouldBeTrue();
@@ -84,6 +90,10 @@ public sealed class MTConnectPumpGenerationTests
             blocking.Set();
             await driver.ShutdownAsync(CancellationToken.None);
         }
+
+        // Shutdown joined every retired pump and forgot them.
+        retired.ShouldAllBe(t => t.IsCompleted);
+        driver.SupersededPumpTasks.ShouldBeEmpty();
     }
 }
 

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

@@ -287,6 +287,56 @@ public sealed class MqttWriteEncoderTests
                 MqttMode.SparkplugB, Sparkplug("Filler1", DriverDataType.Reference), "ref", out _)
             .ShouldBe(MqttWriteEncodeError.UnsupportedDataType);
 
+    // ---- Birth-typed writes keep the birth's OWN Sparkplug datatype (review R5 MINOR-4) ----------
+
+    private static Payload.Types.Metric EncodeBirth(TahuDataType birth, object value)
+    {
+        var error = MqttWriteEncoder.TryEncode(
+            MqttMode.SparkplugB, Sparkplug("Filler1", DriverDataType.Float64) with { DataTypeAuthored = false }, value,
+            out var encoded, birth.ToDriverDataType(), birth);
+        error.ShouldBe(MqttWriteEncodeError.None);
+        return Payload.Parser.ParseFrom(encoded.Payload).Metrics.Single();
+    }
+
+    [Theory]
+    [InlineData(TahuDataType.Int8, (sbyte)-5)]
+    [InlineData(TahuDataType.Uint8, (byte)200)]
+    [InlineData(TahuDataType.Text, "hello")]
+    [InlineData(TahuDataType.Uuid, "123e4567-e89b-12d3-a456-426614174000")]
+    public void A_birth_typed_write_is_commanded_with_the_births_own_datatype(TahuDataType birth, object value)
+        => ((TahuDataType)EncodeBirth(birth, value).Datatype).ShouldBe(birth);
+
+    [Fact]
+    public void A_birth_typed_bytes_write_carries_bytes_with_the_bytes_datatype()
+    {
+        var metric = EncodeBirth(TahuDataType.Bytes, new byte[] { 1, 2, 3 });
+        ((TahuDataType)metric.Datatype).ShouldBe(TahuDataType.Bytes);
+        metric.BytesValue.ToByteArray().ShouldBe([1, 2, 3]);
+    }
+
+    [Theory]
+    [InlineData(TahuDataType.Int32Array)]
+    [InlineData(TahuDataType.DoubleArray)]
+    [InlineData(TahuDataType.DataSet)]
+    [InlineData(TahuDataType.Template)]
+    public void A_birth_type_with_no_command_encoding_is_refused_never_sent_as_another_type(TahuDataType birth)
+    {
+        MqttWriteEncoder.TryEncode(
+                MqttMode.SparkplugB, Sparkplug("Filler1") with { DataTypeAuthored = false }, 1, out var encoded,
+                birth.ToDriverDataType(), birth)
+            .ShouldBe(MqttWriteEncodeError.UnsupportedDataType);
+        encoded.Payload.ShouldBeNull();
+    }
+
+    [Fact]
+    public void An_authored_type_still_wins_over_the_birth_type()
+    {
+        var def = Sparkplug("Filler1", DriverDataType.Int32) with { DataTypeAuthored = true };
+        MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 7, out var encoded, null, TahuDataType.Text)
+            .ShouldBe(MqttWriteEncodeError.None);
+        ((TahuDataType)Payload.Parser.ParseFrom(encoded.Payload).Metrics.Single().Datatype).ShouldBe(TahuDataType.Int32);
+    }
+
     [Fact]
     public void The_mode_decides_the_shape_not_the_blob()
     {