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);