diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index 25c54975..6d1b921e 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -24,20 +24,24 @@ for the design. |---|---| | `Driver.Mqtt.Contracts` | Config records, enums, `MqttTagDefinition` + the pure `MqttTagDefinitionFactory`, and the one shared `MqttJson.Options`. **Deliberately transport-free** — no MQTTnet reference | | `Driver.Mqtt` | Runtime: `MqttDriver`, `MqttConnection` (connect + reconnect supervisor), `MqttSubscriptionManager`, `LastValueCache`, factory registration, `MqttDriverProbe` | +| `Driver.Mqtt.Transport` | `MqttClientOptionsFactory` — TLS posture, PEM CA-pin chain validator, credentials, protocol mapping. Connection **policy** only; nothing here connects | | `Driver.Mqtt.Browser` | `MqttDriverBrowser` + `MqttBrowseSession` — the AdminUI address picker's `#`-observation session | -> `.Browser` references `.Driver` (not just `.Contracts`) — a **documented, deliberate exception** -> so browse can reuse `MqttConnection.BuildClientOptions` (TLS / CA-pin / credentials / protocol -> mapping). Its cost is pulling `Core` into the AdminUI graph. The csproj carries a boxed warning; -> do not copy the pattern into a new driver's browser. The clean fix — extract the pure -> options builder down into `.Contracts` — is tracked as -> [#512](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/512). +> **`.Browser` depends on the shared leaves only** — `.Contracts` + `.Transport`, never the runtime +> `.Driver`. It used to reference `.Driver` under a documented exception; #512 removed it via two +> extractions (the client-options builder to `.Transport`, and the Sparkplug protocol layer — +> `SparkplugTopic`, `SparkplugCodec`, `RebirthRequester` — to `.Contracts`, beside the Tahu wire +> schema they encode). `MqttBrowserLayeringTests` guards the graph in both directions. +> +> There must stay **exactly one copy** of the TLS/CA-pin logic. A second copy would not break the +> build — it would silently diverge, browse quietly not honouring a pinned CA while the driver does. ## Capability Surface ```csharp public sealed class MqttDriver - : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable, IAsyncDisposable, IDisposable + : IDriver, ITagDiscovery, ISubscribable, IReadable, IWritable, IHostConnectivityProbe, IRediscoverable, + IAsyncDisposable, IDisposable ``` | Capability | Entry point | Notes | @@ -46,12 +50,15 @@ public sealed class MqttDriver | `ITagDiscovery` | `DiscoverAsync` | Replays **only the authored `rawTags`** into an `Mqtt` folder. `SupportsOnlineDiscovery => false`. `RediscoverPolicy` is `Once` in Plain and **`UntilStable` in Sparkplug** (half of what a Sparkplug tag needs — its datatype — is only knowable from a birth) | | `ISubscribable` | `SubscribeAsync` / `OnDataChange` | One SUBSCRIBE per distinct authored topic. **`publishingInterval` is ignored** — MQTT is push-based; this driver neither polls nor throttles | | `IReadable` | `ReadAsync` | Serves the last received/retained value from `LastValueCache`; never throws. A never-seen tag reads `BadWaitingForInitialData` | +| `IWritable` | `WriteAsync` | Publishes to the tag's **`writeTopic`** (Plain) or as an **NCMD/DCMD** command (Sparkplug). One status per request; a tag with no write destination is refused `BadNotWritable`. See [Write-through](#write-through) | | `IHostConnectivityProbe` | `GetHostStatuses` | 1-second poll of `MqttConnection.State`; `HostName` is `"{host}:{port}"` | | `IRediscoverable` | `OnRediscoveryNeeded` | Nothing raises it in Plain mode. **Sparkplug raises it** when a birth for an *authored* scope carries a metric-name set different from the last one seen for that scope — both conditions are anti-storm, see [Rediscovery](#rediscovery-sparkplug) | -**Not implemented:** `IWritable`, `IPerCallHostResolver`, `IAlarmSource`, `IHistoryProvider`. -Materialized nodes are `SecurityClassification.ViewOnly`, `IsAlarm: false`, `IsHistorized: false` — -*"MQTT ingest is one-way … ViewOnly rather than an Operate node whose writes would silently vanish."* +**Not implemented:** `IPerCallHostResolver`, `IAlarmSource`, `IHistoryProvider`. +Materialized nodes are `IsAlarm: false`, `IsHistorized: false`. A node is +`SecurityClassification.Operate` when its tag declares a write destination and +`ViewOnly` when it does not — the browse-time answer comes from the same encoder the write path uses, +so a node can never advertise writability it would then refuse. Driver-type string: **`Mqtt`** (`DriverTypeNames.Mqtt`). @@ -274,6 +281,81 @@ nothing forever, so `OnReconnectedAsync` **throws** when zero filters are grante down rather than swallowing it. A SUBACK failure degrades just the affected RawPaths to `BadCommunicationError`. +## Write-through + +`IWritable` landed with [#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508). Before it, +every MQTT node materialized read-only regardless of what the operator authored. + +| Mode | Destination | Payload | +|---|---|---| +| **Plain** | the tag's `writeTopic` | the value as UTF-8 text (`Raw`/`Scalar`) or a bare JSON literal (`Json`) | +| **Sparkplug B** | `spBv1.0/{group}/DCMD/{node}/{device}` for a device metric, `spBv1.0/{group}/NCMD/{node}` for an edge-node metric | a Sparkplug `Payload` carrying the metric **by name**, with an explicit `Datatype` and no `seq` | + +Both publish at **QoS 0, `retain: false`**. 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. + +### `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 +Plain tag with **no** `writeTopic` is refused `BadNotWritable` — it is not published anywhere, and the +node is discovered `ViewOnly`. + +> This is the whole point of the design. The one outcome worse than read-only nodes is a node that +> *looks* writable, accepts a write, returns `Good`, and publishes nothing. Publishing back onto the +> tag's own read topic would be worse still: it would echo straight back and read as a confirmed +> device value. + +```jsonc +{ + "topic": "factory/oven/temp/state", + "writeTopic": "factory/oven/temp/set", // omit for a read-only tag + "payloadFormat": "Scalar", + "dataType": "Float64" +} +``` + +Sparkplug needs no extra key — a command is addressed by the binding tuple the tag already carries. + +### `Good` means published, not 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 +truth arrives afterwards as the echoed value on the read topic and corrects the node through ordinary +ingest (write-outcome self-correction, [#5](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/5)). + +**Inherited consequence:** this driver can never surface a failed *device* write, exactly as Galaxy +cannot. A publish that itself fails **is** reported `Bad` — the optimism covers what the device did +with the command, never whether it was sent. + +### Status codes + +| Situation | Status | +|---|---| +| Published | `Good` | +| Reference is not an authored tag | `BadNodeIdUnknown` | +| No write destination, or a data type with no command encoding | `BadNotWritable` | +| Value not representable in the tag's declared type | `BadTypeMismatch` | +| Publish failed, or no live broker connection | `BadCommunicationError` | + +One status per request, in request order; one bad request never fails the rest of the batch. + +### `WriteIdempotent` stays default-off + +MQTT gives no write ack and a Sparkplug command is not necessarily idempotent (a pulse, a recipe step, +a counter increment), so the resilience pipeline must not replay one. Operators opt in per tag. + +### Two things to know + +- **`jsonPath` is not applied on write.** It is a read-side extraction from the *read* topic's + document; the write topic is a different topic whose payload shape nothing in the tag config + describes. A `Json` write emits a bare JSON literal. Reconstructing a wrapped document from an + extraction path would be guesswork. +- **NCMD is a shared channel.** The driver publishes its own rebirth requests + (`Node Control/Rebirth`) on `spBv1.0/{group}/NCMD/{node}` — the same topic an edge-node metric write + uses. Anything consuming that topic must match on the metric name. The live gate caught exactly this: + a rebirth NCMD arrived before the write command. DCMD has no such contention. + ## Browse `MqttDriverBrowser` opens a **read-only observation session**: it subscribes at QoS 0 to `#` (or @@ -384,14 +466,13 @@ identical words. See [TestConnectProbes.md](TestConnectProbes.md). Every row is tracked. None is a silent failure — each either refuses in words or logs. +**Closed 2026-07-31:** write-through ([#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508), live-gated), the slashed-metric browse-commit +([#509](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/509)), the stale rebirth affordance ([#510](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/510)), the `.Browser` layering exception +([#512](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/512)), the stray Sparkplug `payloadFormat` ([#513](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/513)) and rebirth on a +not-yet-birthed plant ([#514](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/514)). + | Gap | Issue | Effect | Notes | |---|:---:|---|---| -| **Write-through (`IWritable`)** | [#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508) | Every MQTT node is read-only | Deferred to P3 — NCMD/DCMD + plain publish, optimistic-Good + self-correct on echo | -| **Rediscovery is inert** | [#507](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/507) | A DBIRTH introducing a metric does not change the OPC UA tree; redeploy to pick it up | v3 **platform** gap, not MQTT-specific — nothing subscribes to `OnRediscoveryNeeded` and `DriverHostActor.HandleDiscoveredNodes` hard-returns. Also affects Galaxy / TwinCAT / AbCip. Driver-side decision is log-observable | -| **Rebirth needs a non-empty tree** | [#514](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/514) | `Request rebirth…` cannot be armed on a plant that has not birthed since the window opened | See [Refresh](#refresh). Session side already supports a bare `{group}/{edgeNode}` scope | -| **`_canRebirth` is captured at browse-open** | [#510](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/510) | A session reaped for idleness (2 min) still draws the button, and using it errors | The error is shown, not swallowed; reopen the browser | +| **Rediscovery is advisory, not a rebuild** | [#507](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/507) (closed, superseded) | A DBIRTH introducing a metric does not change the served OPC UA tree; an operator must re-browse the device and commit | **No longer "inert".** `DriverInstanceActor` now subscribes to `OnRediscoveryNeeded` and the raise rides the driver-health snapshot to the AdminUI `/hosts` page as a **re-browse chip**. It deliberately does not inject nodes: v3 authors raw tags through the `/raw` browse-commit flow, and grafting nodes at runtime would materialise ones nobody approved and no deployment artifact records. #507's injection path was deleted rather than revived | | **Primary-host STATE publishing** | [#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511) | `ActAsPrimaryHost` does nothing | Logs a startup warning rather than being silently inert — see the `Sparkplug` sub-object table | -| **A metric whose name contains `/` cannot be browse-committed** | [#509](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/509) | The commit is refused **whole**, in words (`Row N: Name must not contain '/'`) | The derived **tag Name** is a RawPath segment and may not contain `/`, while `metricName` legitimately may. This blocks the spec-mandatory `Node Control/Rebirth`. **Workaround: Manual entry** — author the tag with a `/`-free name and type the slashed `metricName` into the Sparkplug editor, which accepts it. Nothing is silently mis-bound | | **`DataSet` / `Template` / `PropertySet` metrics** | [#515](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/515) | Skipped with a warning | v1-unsupported; never coerced into a guessed type | -| **The tag editor writes `payloadFormat` into a Sparkplug blob** | [#513](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/513) | Cosmetic only | `payloadFormat` is Plain-only; the factory routes on the Sparkplug keys and ignores it. Noise in the blob, not a mis-binding | -| **`.Browser` references `.Driver`, not just `.Contracts`** | [#512](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/512) | A documented layering exception | Taken to reuse `MqttConnection.BuildClientOptions` rather than duplicate TLS/CA-pin logic — see the note at the top of this doc | diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagConfigKeys.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagConfigKeys.cs index 4a11784b..a40e6ead 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagConfigKeys.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagConfigKeys.cs @@ -26,6 +26,15 @@ public static class MqttTagConfigKeys /// The concrete MQTT topic a Plain-mode tag subscribes to. public const string Topic = "topic"; + /// + /// The concrete MQTT topic a Plain-mode tag PUBLISHES to when written (Gitea #508). Deliberately + /// separate from : a tag legitimately reads .../state and writes + /// .../set, and publishing a command onto the tag's own read topic would echo back and + /// read as a confirmed device value. Absent ⇒ the tag is not writable, and a write is refused + /// rather than silently dropped. + /// + public const string WriteTopic = "writeTopic"; + /// The Sparkplug group id — the first segment of the tag's binding tuple. public const string GroupId = "groupId"; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs index 15c3c3eb..6bd86ecd 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs @@ -60,6 +60,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// default. Both factories set it from the key's presence so it /// means the same thing in either mode; nothing on the Plain ingest path reads it. /// +/// +/// The concrete MQTT topic a Plain-mode write publishes to, or when the +/// tag declares no write destination — in which case a write is refused with BadNotWritable +/// rather than silently dropped (Gitea #508). Unused in Sparkplug mode, where the command address is +/// the binding tuple itself (an NCMD/DCMD topic derived from group/edgeNode/device). +/// /// The Sparkplug group id; for a Plain-mode tag. /// The Sparkplug edge-node id; for a Plain-mode tag. /// @@ -82,4 +88,5 @@ public sealed record MqttTagDefinition( string? GroupId = null, string? EdgeNodeId = null, string? DeviceId = null, - string? MetricName = null); + string? MetricName = null, + string? WriteTopic = null); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs index ecdcf121..abceea13 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs @@ -104,7 +104,10 @@ public static class MqttTagDefinitionFactory DataType: dataType, Qos: qos, RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true), - DataTypeAuthored: dataTypeAuthored); + DataTypeAuthored: dataTypeAuthored, + // Optional, and absence is meaningful: no writeTopic ⇒ not writable (Gitea #508). + // Blank normalises to null so "" and absent cannot mean different things. + WriteTopic: NullIfBlank(ReadString(root, MqttTagConfigKeys.WriteTopic))); return true; } catch (JsonException) { return false; } @@ -190,7 +193,10 @@ public static class MqttTagDefinitionFactory GroupId: groupId, EdgeNodeId: edgeNodeId, DeviceId: deviceId, - MetricName: metricName); + MetricName: metricName, + // Carried through, but unread on the Sparkplug write path: a Sparkplug command is + // addressed by the binding tuple, not by a topic the operator typed. + WriteTopic: NullIfBlank(ReadString(root, MqttTagConfigKeys.WriteTopic))); return true; } catch (JsonException) { return false; } @@ -283,6 +289,17 @@ public static class MqttTagDefinitionFactory ? e.GetString() ?? "" : ""; + /// + /// Normalises a blank string to , for the optional fields where absence is + /// itself meaningful — writeTopic, where "absent" means "not writable" (Gitea #508). Without + /// this, "writeTopic": "" and an absent key would take different code paths and an empty + /// string could reach the transport as a publish topic. + /// + /// The raw read value. + /// The trimmed value, or when blank. + private static string? NullIfBlank(string value) + => string.IsNullOrWhiteSpace(value) ? null : value.Trim(); + private static bool ReadBoolOrDefault(JsonElement o, string name, bool defaultValue) => o.TryGetProperty(name, out var e) && e.ValueKind is JsonValueKind.True or JsonValueKind.False ? e.GetBoolean() 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 new file mode 100644 index 00000000..45ee2981 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs @@ -0,0 +1,304 @@ +using System.Globalization; +using System.Text; +using System.Text.Json; +using Google.Protobuf; +using Org.Eclipse.Tahu.Protobuf; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; + +/// +/// Why a write could not be encoded. Maps 1:1 onto the status the driver returns, so the reason a +/// write was refused is decided here — once, purely — rather than at the transport. +/// +public enum MqttWriteEncodeError +{ + /// Encoding succeeded. + None = 0, + + /// + /// The tag declares no write destination: a Plain tag with no writeTopic. Deliberately a + /// hard refusal — see 's remarks on why a silent no-op would be + /// worse than the honest read-only behaviour this replaced. + /// + NoWriteDestination, + + /// The value is null, or cannot be represented in the tag's declared data type. + ValueNotRepresentable, + + /// The tag's declared data type has no Sparkplug command encoding. + UnsupportedDataType, +} + +/// +/// Encodes an outbound MQTT write — the payload bytes and the topic to publish them on — for both +/// ingest shapes (Gitea #508). Pure: no I/O, no transport, no clock beyond the Sparkplug command +/// timestamp, so the whole encode contract is unit-testable without a broker. +/// +/// +/// +/// A missing write destination is a refusal, never a no-op. Before #508 the driver +/// implemented no IWritable at all, so every MQTT node materialised read-only — honest, if +/// limited. The one outcome that would be strictly worse is a node that looks writable, +/// accepts a write, returns Good and publishes nothing. So a Plain tag with no writeTopic +/// yields and the driver surfaces +/// BadNotWritable. +/// +/// +/// The write topic is deliberately separate from the read topic. A tag legitimately reads +/// .../state and writes .../set, so the destination is its own authored field rather +/// than being inferred from . Publishing a command back onto +/// the tag's own read topic would, on a broker with no device behind it, echo straight back and +/// read as a confirmed device value. +/// +/// +/// Sparkplug writes are commands, not data. A metric under a device encodes as a DCMD, one +/// owned by the edge node itself as an NCMD, addressed by the tag's stable +/// (group, edgeNode, device?, metric) tuple — never by the per-birth alias, which may be +/// reused for a different metric across a rebirth. No seq is stamped, for the reason +/// documents at length: the sequence stream belongs to the messages +/// an edge node itself publishes, and a strict node validating a host-supplied one against its own +/// counter would have grounds to reject the command. +/// +/// +/// jsonPath is not applied on write. It is a read-side extraction from the read topic's +/// document, and the write topic is a different topic whose payload shape nothing in the tag config +/// describes. A write therefore emits the value as a bare JSON +/// literal. Reconstructing a wrapped document from an extraction path would be guesswork, and +/// guessing a payload shape is how a write lands somewhere nobody authored. +/// +/// +public static class MqttWriteEncoder +{ + /// QoS for an outbound write. See the remarks on . + public const int WriteQos = 0; + + /// + /// Encodes one write for the tag under the driver's . + /// + /// The driver's ingest shape — chosen by driver config, never sniffed from the tag. + /// The resolved tag definition. + /// The value to write. + /// The topic + payload to publish, when this returns . + /// The encode outcome. + /// + /// QoS 0, retain . 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 + /// configurable. + /// + public static MqttWriteEncodeError TryEncode( + MqttMode mode, + MqttTagDefinition def, + object? value, + out MqttEncodedWrite encoded) + { + ArgumentNullException.ThrowIfNull(def); + encoded = default; + + return mode == MqttMode.SparkplugB + ? TryEncodeSparkplug(def, value, out encoded) + : TryEncodePlain(def, value, out encoded); + } + + /// Encodes a Plain-mode write: the value's text/JSON rendering onto the tag's write topic. + /// The resolved tag definition. + /// The value to write. + /// The topic + payload on success. + /// The encode outcome. + private static MqttWriteEncodeError TryEncodePlain( + MqttTagDefinition def, + object? value, + out MqttEncodedWrite encoded) + { + encoded = default; + + if (string.IsNullOrWhiteSpace(def.WriteTopic)) + { + return MqttWriteEncodeError.NoWriteDestination; + } + + if (value is null) + { + // No MQTT payload means "the value is null" — a zero-length payload is a legitimate + // clear-the-retained-message instruction, which is a different operation entirely. + return MqttWriteEncodeError.ValueNotRepresentable; + } + + string text; + if (def.PayloadFormat == MqttPayloadFormat.Json) + { + try + { + text = JsonSerializer.Serialize(value); + } + catch (NotSupportedException) + { + return MqttWriteEncodeError.ValueNotRepresentable; + } + } + else + { + // Raw and Scalar are both "the payload's UTF-8 text" on the read side; the write side + // mirrors that. Invariant culture, or a comma decimal separator on a European host would + // publish a number the device cannot parse. + text = value is IFormattable f + ? f.ToString(null, CultureInfo.InvariantCulture) + : value.ToString() ?? ""; + } + + encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text)); + return MqttWriteEncodeError.None; + } + + /// Encodes a Sparkplug-mode write as an NCMD (node metric) or DCMD (device metric). + /// The resolved tag definition. + /// The value to write. + /// The topic + payload on success. + /// The encode outcome. + private static MqttWriteEncodeError TryEncodeSparkplug( + MqttTagDefinition def, + object? value, + out MqttEncodedWrite encoded) + { + encoded = default; + + if (def.GroupId is not { Length: > 0 } groupId + || def.EdgeNodeId is not { Length: > 0 } edgeNodeId + || def.MetricName is not { Length: > 0 } metricName) + { + // A Sparkplug tag's command address IS its binding tuple, so an incomplete tuple is the + // Sparkplug equivalent of a missing write topic rather than a different kind of failure. + return MqttWriteEncodeError.NoWriteDestination; + } + + if (value is null) + { + return MqttWriteEncodeError.ValueNotRepresentable; + } + + var metric = new Payload.Types.Metric { Name = metricName }; + var error = ApplyValue(metric, def.DataType, value); + if (error != MqttWriteEncodeError.None) + { + return error; + } + + var timestampMs = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + metric.Timestamp = timestampMs; + + var payload = new Payload { Timestamp = timestampMs }; + payload.Metrics.Add(metric); + + // A device metric is commanded with DCMD addressed to that device; an edge-node metric with + // NCMD. Sending a device metric as NCMD would address the node, which does not own it. + var isDevice = def.DeviceId is { Length: > 0 }; + var topic = SparkplugTopic.Format( + groupId, + isDevice ? SparkplugMessageType.DCMD : SparkplugMessageType.NCMD, + edgeNodeId, + isDevice ? def.DeviceId : null); + + encoded = new MqttEncodedWrite(topic, payload.ToByteArray()); + return MqttWriteEncodeError.None; + } + + /// + /// Sets the metric's typed value field and its Datatype from the tag's declared type. + /// + /// The metric being built. + /// The tag's declared driver data type. + /// The value to convert. + /// The encode outcome. + /// + /// Datatype is set explicitly rather than left for the node to infer, for the same reason + /// sets it: a node that treats a missing or ambiguous field as + /// malformed drops the command silently, and a silently-dropped write is the failure mode this + /// whole feature must not have. + /// + private static MqttWriteEncodeError ApplyValue( + Payload.Types.Metric metric, + DriverDataType dataType, + object value) + { + try + { + switch (dataType) + { + case DriverDataType.Boolean: + metric.Datatype = (uint)TahuDataType.Boolean; + metric.BooleanValue = Convert.ToBoolean(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.Int16: + metric.Datatype = (uint)TahuDataType.Int16; + metric.IntValue = unchecked((uint)Convert.ToInt16(value, CultureInfo.InvariantCulture)); + return MqttWriteEncodeError.None; + + case DriverDataType.Int32: + metric.Datatype = (uint)TahuDataType.Int32; + metric.IntValue = unchecked((uint)Convert.ToInt32(value, CultureInfo.InvariantCulture)); + return MqttWriteEncodeError.None; + + case DriverDataType.Int64: + metric.Datatype = (uint)TahuDataType.Int64; + metric.LongValue = unchecked((ulong)Convert.ToInt64(value, CultureInfo.InvariantCulture)); + return MqttWriteEncodeError.None; + + case DriverDataType.UInt16: + metric.Datatype = (uint)TahuDataType.Uint16; + metric.IntValue = Convert.ToUInt16(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.UInt32: + metric.Datatype = (uint)TahuDataType.Uint32; + metric.IntValue = Convert.ToUInt32(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.UInt64: + metric.Datatype = (uint)TahuDataType.Uint64; + metric.LongValue = Convert.ToUInt64(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.Float32: + metric.Datatype = (uint)TahuDataType.Float; + metric.FloatValue = Convert.ToSingle(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.Float64: + metric.Datatype = (uint)TahuDataType.Double; + metric.DoubleValue = Convert.ToDouble(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.String: + metric.Datatype = (uint)TahuDataType.String; + metric.StringValue = value as string ?? Convert.ToString(value, CultureInfo.InvariantCulture) ?? ""; + return MqttWriteEncodeError.None; + + case DriverDataType.DateTime: + metric.Datatype = (uint)TahuDataType.DateTime; + metric.LongValue = (ulong)new DateTimeOffset( + Convert.ToDateTime(value, CultureInfo.InvariantCulture).ToUniversalTime()) + .ToUnixTimeMilliseconds(); + return MqttWriteEncodeError.None; + + default: + // DriverDataType.Reference and anything added later: refuse rather than guess a + // wire representation. An unsupported type must not become a String on the wire. + return MqttWriteEncodeError.UnsupportedDataType; + } + } + catch (Exception ex) when (ex is InvalidCastException or FormatException or OverflowException) + { + return MqttWriteEncodeError.ValueNotRepresentable; + } + } +} + +/// 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); 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 3fd63a76..4f1e4d54 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs @@ -63,8 +63,31 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// /// public sealed class MqttDriver - : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable, IAsyncDisposable, IDisposable + : IDriver, ITagDiscovery, ISubscribable, IReadable, IWritable, IHostConnectivityProbe, IRediscoverable, + IAsyncDisposable, IDisposable { + // ---- Write status codes (Gitea #508) ---- + // + // NAMED constants, never inline literals: StatusCodeParityTests checks each against the pinned SDK + // by reflection and can only see a named const (Gitea #497/#519). The names must match + // Opc.Ua.StatusCodes members exactly. + + /// The reference is not an authored tag of this driver's live ingest path. + private const uint StatusBadNodeIdUnknown = 0x80340000u; + + /// + /// The tag declares no write destination — a Plain tag with no writeTopic, or a Sparkplug + /// tag with an incomplete binding tuple. The honest answer to "can I write this?", and the whole + /// reason the encoder refuses rather than no-opping. + /// + private const uint StatusBadNotWritable = 0x803B0000u; + + /// The supplied value cannot be represented in the tag's declared data type. + private const uint StatusBadTypeMismatch = 0x80740000u; + + /// The publish failed, or the driver has no live broker connection to publish through. + private const uint StatusBadCommunicationError = 0x80050000u; + /// /// Rough per-tag driver-attributable cost: the parsed (five /// strings — RawPath, topic, JSONPath and the record header) plus its cached @@ -480,11 +503,16 @@ public sealed class MqttDriver DriverDataType: ResolveDiscoveredDataType(def), IsArray: false, ArrayDim: null, - // MQTT ingest is one-way in P1 — this driver implements no IWritable leg, so every - // node is ViewOnly rather than an Operate node whose writes would silently vanish. - SecurityClass: SecurityClassification.ViewOnly, + // Gitea #508: MQTT now has a write leg, so a tag that declares a write destination is + // an Operate node. One that does not stays ViewOnly — the point of the distinction is + // that a node must never look writable while its writes would go nowhere. + SecurityClass: CanWrite(def) ? SecurityClassification.Operate : SecurityClassification.ViewOnly, IsHistorized: false, IsAlarm: false, + // Default-off, deliberately: MQTT gives no application-level write acknowledgement, and + // a Sparkplug command is not necessarily idempotent (a pulse, a recipe step, a counter + // increment). The resilience pipeline must not replay one by default; an operator opts + // in per tag via Tag.WriteIdempotent. WriteIdempotent: false)); } @@ -585,6 +613,142 @@ public sealed class MqttDriver return Task.FromResult>(results); } + // ---- IWritable (Gitea #508) ---- + + /// + /// Whether declares somewhere for a write to go: a writeTopic in + /// Plain mode, or a complete command tuple in Sparkplug mode. + /// + /// The resolved tag definition. + /// when a write to this tag can be addressed. + /// + /// Asked of the SAME encoder the write path uses, so the browse-time answer and the runtime + /// answer cannot drift — a node advertised Operate that then refuses every write would be the + /// silent-no-op failure in a new costume. The value is irrelevant here, so a probe value of the + /// tag's own type is encoded and thrown away. + /// + private bool CanWrite(MqttTagDefinition def) + => MqttWriteEncoder.TryEncode(_options.Mode, def, ProbeValueFor(def.DataType), out _) + != MqttWriteEncodeError.NoWriteDestination; + + /// A never-published sentinel value of the given type, for the probe. + /// The tag's declared data type. + /// A representable value, or a string for types with no natural zero. + private static object ProbeValueFor(DriverDataType dataType) => dataType switch + { + DriverDataType.Boolean => false, + DriverDataType.DateTime => DateTime.UnixEpoch, + DriverDataType.String or DriverDataType.Reference => "", + _ => 0, + }; + + /// + /// Publishes a batch of writes: one MQTT message per request, to the tag's write topic (Plain) + /// or as an NCMD/DCMD command (Sparkplug B). + /// + /// The requested writes. + /// Cancellation for the batch. + /// One result per request, in request order. + /// + /// + /// Never throws, and one bad request never fails the batch. Each slot carries its own + /// status, matching : failing the whole call would let a single stale + /// reference reject every other write in the same OPC UA request. + /// + /// + /// Good means "published", not "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 + /// through the ordinary ingest path. A write whose PUBLISH itself fails is reported Bad: the + /// optimism covers what the device did with the command, never whether it was sent. + /// + /// + /// Consequence, inherited knowingly (see #5, write-outcome self-correction): because the + /// transport cannot report device-level rejection, this driver can never surface a failed + /// device write — exactly the property Galaxy has. A device that silently ignores a + /// command leaves the node reading whatever it last published, which is the honest picture; a + /// device that has no echo topic at all leaves the operator with no confirmation, which is a + /// property of the plant's topic design rather than of this code. + /// + /// + public async Task> WriteAsync( + IReadOnlyList writes, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(writes); + + var results = new WriteResult[writes.Count]; + var connection = _connection; + + for (var i = 0; i < writes.Count; i++) + { + var w = writes[i]; + + if (!TryResolve(w.FullReference, out var def)) + { + results[i] = new WriteResult(StatusBadNodeIdUnknown); + continue; + } + + var error = MqttWriteEncoder.TryEncode(_options.Mode, def, w.Value, out var encoded); + if (error != MqttWriteEncodeError.None) + { + results[i] = new WriteResult(MapEncodeError(error)); + continue; + } + + if (connection is null) + { + // Not yet connected, or torn down mid-rebuild. Reported rather than queued: a write + // buffered now and flushed at an unbounded later moment is the replay hazard + // WriteIdempotent exists to prevent, decided by the driver instead of the operator. + results[i] = new WriteResult(StatusBadCommunicationError); + continue; + } + + try + { + await connection + .PublishAsync(encoded.Topic, encoded.Payload, MqttWriteEncoder.WriteQos, retain: false, cancellationToken) + .ConfigureAwait(false); + results[i] = new WriteResult(0u); + } + 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; + } + catch (Exception ex) + { + _logger?.LogWarning( + ex, + "MQTT driver '{DriverId}': write to '{Reference}' could not be published on '{Topic}'.", + _driverInstanceId, + w.FullReference, + encoded.Topic); + results[i] = new WriteResult(StatusBadCommunicationError); + } + } + + return results; + } + + /// Maps an encode refusal to the status an OPC UA client sees. + /// The encode outcome. + /// The OPC UA status code. + private static uint MapEncodeError(MqttWriteEncodeError error) => error switch + { + MqttWriteEncodeError.NoWriteDestination => StatusBadNotWritable, + MqttWriteEncodeError.UnsupportedDataType => StatusBadNotWritable, + _ => StatusBadTypeMismatch, + }; + // ---- IHostConnectivityProbe ---- /// The single broker this driver instance talks to, as host:port. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor index 456c1626..39ecb1e0 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor @@ -30,6 +30,18 @@ @onchange="@(e => Update(() => _m.Topic = e.Value?.ToString() ?? ""))" />
Concrete topic — MQTT wildcards (+ / #) are not valid for a tag.
+
+ + +
+ Where a write is published. Deliberately separate from the read topic — a tag commonly + reads …/state and writes …/set. Leave blank for a + read-only tag: a write is then refused (BadNotWritable) rather than published + nowhere. +
+