From 1f4f4fca8f3fddd15972536f0692e0477fec3f29 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 07:27:42 -0400 Subject: [PATCH] fix(mqtt): birth-typed Sparkplug writes and QoS 1 for Plain writes (#602 STAB-25/26) Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- .../MqttWriteEncoder.cs | 37 +++++++++++--- .../MqttDriver.cs | 15 ++++-- .../MqttWriteEncoderTests.cs | 49 +++++++++++++++++++ 3 files changed, 90 insertions(+), 11 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs index 6d0b1125..5ebd5d96 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs @@ -74,9 +74,20 @@ public enum MqttWriteEncodeError /// public static class MqttWriteEncoder { - /// QoS for an outbound write. See the remarks on . + /// + /// QoS for a Sparkplug NCMD/DCMD command. See the remarks on . Plain-mode + /// writes use instead. + /// public const int WriteQos = 0; + /// + /// QoS for a Plain-mode writeTopic publish (STAB-26). At QoS 1 the publish completes only + /// once the broker has PUBACKed (or refused) it, so a Good write status means "the broker accepted + /// the message" rather than "the bytes left this socket"; at QoS 0 a broker-side refusal or a + /// connection that dropped under the send was indistinguishable from success. + /// + public const int PlainWriteQos = 1; + /// /// Encodes one write for the tag under the driver's . /// @@ -84,9 +95,16 @@ public static class MqttWriteEncoder /// The resolved tag definition. /// The value to write. /// The topic + payload to publish, when this returns . + /// + /// Sparkplug only (STAB-25): the type to encode the metric as when the tag did not author one — the + /// type the metric's live birth certificate declared, the same authored ? authored : birth + /// precedence the ingest path coerces reads with. encodes as + /// , 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. + /// /// The encode outcome. /// - /// QoS 0, retain . Retain is the one that matters: a retained + /// Retain always; QoS 1 for Plain, 0 for Sparkplug commands. Retain is the one that matters: a retained /// command would be replayed by the broker to the device on every future subscribe — including /// the device's own reconnect — so a single stale write would be re-applied forever. This is the /// same hazard avoids for NCMD, and the reason neither is @@ -96,13 +114,14 @@ public static class MqttWriteEncoder MqttMode mode, MqttTagDefinition def, object? value, - out MqttEncodedWrite encoded) + out MqttEncodedWrite encoded, + DriverDataType? effectiveDataType = null) { ArgumentNullException.ThrowIfNull(def); encoded = default; return mode == MqttMode.SparkplugB - ? TryEncodeSparkplug(def, value, out encoded) + ? TryEncodeSparkplug(def, value, effectiveDataType, out encoded) : TryEncodePlain(def, value, out encoded); } @@ -163,7 +182,7 @@ public static class MqttWriteEncoder : value.ToString() ?? ""; } - encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text)); + encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text), PlainWriteQos); return MqttWriteEncodeError.None; } @@ -175,6 +194,7 @@ public static class MqttWriteEncoder private static MqttWriteEncodeError TryEncodeSparkplug( MqttTagDefinition def, object? value, + DriverDataType? effectiveDataType, out MqttEncodedWrite encoded) { encoded = default; @@ -206,7 +226,7 @@ public static class MqttWriteEncoder } var metric = new Payload.Types.Metric { Name = metricName }; - var error = ApplyValue(metric, def.DataType, value); + var error = ApplyValue(metric, def.DataTypeAuthored ? def.DataType : effectiveDataType ?? def.DataType, value); if (error != MqttWriteEncodeError.None) { return error; @@ -227,7 +247,7 @@ public static class MqttWriteEncoder edgeNodeId, isDevice ? def.DeviceId : null); - encoded = new MqttEncodedWrite(topic, payload.ToByteArray()); + encoded = new MqttEncodedWrite(topic, payload.ToByteArray(), WriteQos); return MqttWriteEncodeError.None; } @@ -344,4 +364,5 @@ public static class MqttWriteEncoder /// One encoded outbound write: where it goes and what it carries. /// The concrete topic to publish on. /// The encoded message body. -public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload); +/// The MQTT QoS to publish at — see / . +public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload, int Qos = MqttWriteEncoder.WriteQos); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs index 306ea3c8..36f5182b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs @@ -544,6 +544,11 @@ public sealed class MqttDriver /// /// /// + /// Also the write path's type (STAB-25). encodes a Sparkplug + /// command with this same answer, so a birth-typed Int32 metric is commanded as Int32 rather than + /// as the record's default. + /// + /// /// The precedence is the ingest path's, restated. SparkplugIngestor.PublishMetric /// coerces every value with DataTypeAuthored ? authored : birth; a discovery surface /// that disagreed with it would advertise a type no value ever arrives as. The two must be @@ -674,7 +679,10 @@ public sealed class MqttDriver /// reference reject every other write in the same OPC UA request. /// /// - /// Good means "published", not "applied". MQTT has no application-level write + /// Good means "published", not "applied". A Plain-mode write is published at QoS 1 + /// (STAB-26), so "published" means the broker PUBACKed it — a broker-side refusal or a drop under + /// the send is Bad, not Good. Sparkplug NCMD/DCMD stay at QoS 0 (a redelivered command would be + /// re-applied). MQTT has no application-level write /// acknowledgement, so a successful PUBLISH is the strongest truth available at this seam — /// the same optimistic-Good posture Galaxy's fire-and-forget write takes. Device-level truth /// arrives afterwards, as the echoed value on the tag's read topic, and corrects the node @@ -709,7 +717,8 @@ public sealed class MqttDriver continue; } - var error = MqttWriteEncoder.TryEncode(_options.Mode, def, w.Value, out var encoded); + var error = MqttWriteEncoder.TryEncode( + _options.Mode, def, w.Value, out var encoded, ResolveDiscoveredDataType(def)); if (error != MqttWriteEncodeError.None) { results[i] = new WriteResult(MapEncodeError(error)); @@ -728,7 +737,7 @@ public sealed class MqttDriver try { await connection - .PublishAsync(encoded.Topic, encoded.Payload, MqttWriteEncoder.WriteQos, retain: false, cancellationToken) + .PublishAsync(encoded.Topic, encoded.Payload, encoded.Qos, retain: false, cancellationToken) .ConfigureAwait(false); results[i] = new WriteResult(0u); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs index 20984a5b..56784606 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs @@ -350,6 +350,55 @@ public sealed class MqttWriteEncoderTests Payload.Parser.ParseFrom(encoded.Payload).Metrics[0].Name.ShouldBe("Node Control/Rebirth"); } + // ---- STAB-25 / STAB-26 (#602) --------------------------------------------------------------------- + + [Fact] + public void A_plain_write_publishes_at_QoS1_so_Good_means_the_broker_accepted_it() + { + MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("oven/set"), 42.5, out var encoded) + .ShouldBe(MqttWriteEncodeError.None); + + encoded.Qos.ShouldBe(1); + } + + [Fact] + public void A_sparkplug_command_stays_at_QoS0() + { + // Pinned beside the Plain case so the two cannot be "unified" by accident: an NCMD/DCMD retry from + // QoS 1 redelivery would re-apply a command the device already took. + MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, Sparkplug("Filler1"), 1.0, out var encoded) + .ShouldBe(MqttWriteEncodeError.None); + + encoded.Qos.ShouldBe(0); + } + + [Fact] + public void An_unauthored_sparkplug_type_is_commanded_with_the_birth_declared_type() + { + // DataTypeAuthored=false: the record's Float64 is only a default. The birth said Int32, so the DCMD + // must carry Int32 — a Double-typed command to an Int32 metric is the silent-drop shape. + var def = Sparkplug("Filler1") with { DataTypeAuthored = false }; + + MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 7, out var encoded, DriverDataType.Int32) + .ShouldBe(MqttWriteEncodeError.None); + + var metric = Payload.Parser.ParseFrom(encoded.Payload).Metrics[0]; + metric.Datatype.ShouldBe((uint)TahuDataType.Int32); + metric.IntValue.ShouldBe(7u); + } + + [Fact] + public void An_authored_sparkplug_type_wins_over_the_birth_type() + { + // Same precedence as the ingest path: authored beats birth. + var def = Sparkplug("Filler1", DriverDataType.Float64) with { DataTypeAuthored = true }; + + MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 7, out var encoded, DriverDataType.Int32) + .ShouldBe(MqttWriteEncodeError.None); + + Payload.Parser.ParseFrom(encoded.Payload).Metrics[0].Datatype.ShouldBe((uint)TahuDataType.Double); + } + private static string EncodeTopic(MqttMode mode, MqttTagDefinition def) { MqttWriteEncoder.TryEncode(mode, def, 1.0, out var encoded).ShouldBe(MqttWriteEncodeError.None);