merge: MQTT tracker batch — #508, #509, #510, #512, #513, #514
v2-ci / build (push) Successful in 4m12s
v2-ci / unit-tests (push) Failing after 13m35s

Write-through (live-gated against the real Mosquitto fixture), four AdminUI
authoring defects, and the .Browser layering exception.

Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
This commit is contained in:
Joseph Doherty
2026-07-31 06:42:12 -04:00
32 changed files with 2470 additions and 293 deletions
+1
View File
@@ -49,6 +49,7 @@
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.csproj" />
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj" />
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj" />
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport.csproj" />
<Project Path="src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj" />
</Folder>
<Folder Name="/src/Drivers/Driver CLIs/">
+98 -17
View File
@@ -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 |
@@ -23,12 +23,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;
/// sanctioned publish, and it is never reached by <c>OpenAsync</c> or by any browse call.
/// </para>
/// <para>
/// <b>Layering note.</b> Unlike every other bespoke browser here, this project references the
/// runtime <c>.Driver</c> project (to reuse <c>MqttConnection.BuildClientOptions</c> rather
/// than fork the TLS / CA-pin path). That is a known, deliberate exception with a documented
/// cost and a documented clean fix — see the comment on that <c>ProjectReference</c> in
/// <c>ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj</c> before copying this project as a
/// template.
/// <b>Layering note.</b> This project depends on the shared leaves only — <c>.Contracts</c> for
/// the Sparkplug wire schema and codec, <c>.Transport</c> for
/// <see cref="MqttClientOptionsFactory"/> — and NOT on the runtime <c>.Driver</c> project, which
/// it used to reference under a documented exception (Gitea #512). Reuse of the TLS / CA-pin path
/// is what that exception bought, and it is preserved: there is still exactly one copy, now in
/// <c>.Transport</c>. Do not fork it. Pinned by <c>MqttBrowserLayeringTests</c>.
/// </para>
/// </summary>
public sealed class MqttDriverBrowser : IDriverBrowser
@@ -73,7 +73,7 @@ public sealed class MqttDriverBrowser : IDriverBrowser
var filter = BuildRootFilter(opts);
var suffix = BuildBrowseClientIdSuffix();
var clientOptions = MqttConnection.BuildClientOptions(ToBrowseOptions(opts), suffix, _logger);
var clientOptions = MqttClientOptionsFactory.BuildClientOptions(ToBrowseOptions(opts), suffix, _logger);
using var openCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
openCts.CancelAfter(OpenBudget(opts));
@@ -14,34 +14,36 @@
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Commons\ZB.MOM.WW.OtOpcUa.Commons.csproj"/>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
<!--
┌──────────────────────────────────────────────────────────────────────────────────┐
│ KNOWN, DELIBERATE EXCEPTION to the .Browser → .Contracts pattern. │
│ DO NOT COPY THIS LINE INTO A NEW DRIVER'S BROWSER WITHOUT READING THIS. │
└──────────────────────────────────────────────────────────────────────────────────┘
The layering exception this project used to carry is GONE (Gitea #512). It referenced the
runtime .Driver project, and that reference reached the AdminUI, which references this
browser: the AdminUI's build/deploy graph pulled in the whole MQTT driver runtime.
Every other bespoke browser in this repo (OpcUaClient, Galaxy) references only its own
.Contracts project plus its transport package. This one also references the runtime
.Driver project. That is a real widening, not a free one, and it was accepted knowingly:
MEASURED, not assumed: after the change ZB.MOM.WW.OtOpcUa.Driver.Mqtt.dll, Polly.Core and
Serilog are all absent from the AdminUI's output. ZB.MOM.WW.OtOpcUa.Core.dll is still there,
because Driver.Sql.Browser ALSO references its runtime driver — the identical violation, in a
second driver, and undocumented (this one at least carried a comment). The comment that used
to live here claimed this browser was the sole cause of Core reaching the AdminUI; it was not.
Tracked separately rather than fixed in passing.
WHY: the browse CONNECT reuses MqttConnection.BuildClientOptions, which owns the TLS
posture, the PEM CA-pin chain validator, the credential handling and the protocol
version mapping (~100 lines). A second copy here would be a silent security
divergence — browse quietly not honouring a pinned CA while the runtime driver does.
The method is pure (no I/O, no network), so the dependency costs nothing at run time.
It took TWO extractions to remove, not the one the issue anticipated:
WHAT IT COSTS: the AdminUI's project graph does NOT otherwise include Core.csproj
(only Host.csproj and the runtime Driver.* projects do). Referencing this browser
from the AdminUI therefore pulls in Core + Polly.Core + Serilog at build/deploy time.
1. MqttClientOptionsFactory (TLS posture, PEM CA-pin chain validator, credentials,
protocol-version mapping) moved to the leaf .Transport project below — the fix the
comment that used to live here had already prescribed. It cannot live in .Contracts:
that project is deliberately transport-free, and BuildClientOptions returns
MqttClientOptions, an MQTTnet type.
THE CLEAN FIX (deferred, not blocked): a small leaf project — say
ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport (net10, refs .Contracts + MQTTnet) — holding
BuildClientOptions/ConfigureTls/the CA-pin helpers, referenced by BOTH .Driver and
.Browser. Note the obvious-looking alternative does NOT work: those members cannot
move into .Contracts, because .Contracts is deliberately transport-free (Task 1
removed its MQTTnet reference to keep it so) and BuildClientOptions returns
MqttClientOptions, an MQTTnet type.
2. The Sparkplug PROTOCOL layer — SparkplugTopic/SparkplugMessageType, SparkplugCodec and
RebirthRequester — moved to .Contracts, which is where the Tahu wire schema they
encode/decode already lived. This half is easy to miss: the browse session decodes
births itself, so extracting only the options builder would have left the .Driver
reference exactly where it was.
Whatever else changes here, there must stay exactly ONE copy of the TLS/CA-pin logic. The
failure mode of a second copy is not a build break but a silent security divergence — browse
quietly not honouring a pinned CA while the runtime driver does.
-->
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport.csproj"/>
</ItemGroup>
<ItemGroup>
@@ -26,6 +26,15 @@ public static class MqttTagConfigKeys
/// <summary>The concrete MQTT topic a Plain-mode tag subscribes to.</summary>
public const string Topic = "topic";
/// <summary>
/// The concrete MQTT topic a Plain-mode tag PUBLISHES to when written (Gitea #508). Deliberately
/// separate from <see cref="Topic"/>: a tag legitimately reads <c>.../state</c> and writes
/// <c>.../set</c>, 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.
/// </summary>
public const string WriteTopic = "writeTopic";
/// <summary>The Sparkplug group id — the first segment of the tag's binding tuple.</summary>
public const string GroupId = "groupId";
@@ -60,6 +60,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// <see cref="DriverDataType.String"/> 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.
/// </param>
/// <param name="WriteTopic">
/// The concrete MQTT topic a <b>Plain</b>-mode write publishes to, or <see langword="null"/> when the
/// tag declares no write destination — in which case a write is refused with <c>BadNotWritable</c>
/// 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).
/// </param>
/// <param name="GroupId">The Sparkplug group id; <see langword="null"/> for a Plain-mode tag.</param>
/// <param name="EdgeNodeId">The Sparkplug edge-node id; <see langword="null"/> for a Plain-mode tag.</param>
/// <param name="DeviceId">
@@ -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);
@@ -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() ?? ""
: "";
/// <summary>
/// Normalises a blank string to <see langword="null"/>, for the optional fields where absence is
/// itself meaningful — <c>writeTopic</c>, where "absent" means "not writable" (Gitea #508). Without
/// this, <c>"writeTopic": ""</c> and an absent key would take different code paths and an empty
/// string could reach the transport as a publish topic.
/// </summary>
/// <param name="value">The raw read value.</param>
/// <returns>The trimmed value, or <see langword="null"/> when blank.</returns>
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()
@@ -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;
/// <summary>
/// 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.
/// </summary>
public enum MqttWriteEncodeError
{
/// <summary>Encoding succeeded.</summary>
None = 0,
/// <summary>
/// The tag declares no write destination: a Plain tag with no <c>writeTopic</c>. Deliberately a
/// hard refusal — see <see cref="MqttWriteEncoder"/>'s remarks on why a silent no-op would be
/// worse than the honest read-only behaviour this replaced.
/// </summary>
NoWriteDestination,
/// <summary>The value is null, or cannot be represented in the tag's declared data type.</summary>
ValueNotRepresentable,
/// <summary>The tag's declared data type has no Sparkplug command encoding.</summary>
UnsupportedDataType,
}
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// <para>
/// <b>A missing write destination is a refusal, never a no-op.</b> Before #508 the driver
/// implemented no <c>IWritable</c> at all, so every MQTT node materialised read-only — honest, if
/// limited. The one outcome that would be strictly worse is a node that <em>looks</em> writable,
/// accepts a write, returns Good and publishes nothing. So a Plain tag with no <c>writeTopic</c>
/// yields <see cref="MqttWriteEncodeError.NoWriteDestination"/> and the driver surfaces
/// <c>BadNotWritable</c>.
/// </para>
/// <para>
/// <b>The write topic is deliberately separate from the read topic.</b> A tag legitimately reads
/// <c>.../state</c> and writes <c>.../set</c>, so the destination is its own authored field rather
/// than being inferred from <see cref="MqttTagDefinition.Topic"/>. 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.
/// </para>
/// <para>
/// <b>Sparkplug writes are commands, not data.</b> 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
/// <c>(group, edgeNode, device?, metric)</c> tuple — never by the per-birth alias, which may be
/// reused for a different metric across a rebirth. No <c>seq</c> is stamped, for the reason
/// <see cref="RebirthRequester"/> 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.
/// </para>
/// <para>
/// <b>jsonPath is not applied on write.</b> 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 <see cref="MqttPayloadFormat.Json"/> 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.
/// </para>
/// </remarks>
public static class MqttWriteEncoder
{
/// <summary>QoS for an outbound write. See the remarks on <see cref="TryEncode"/>.</summary>
public const int WriteQos = 0;
/// <summary>
/// Encodes one write for the tag <paramref name="def"/> under the driver's <paramref name="mode"/>.
/// </summary>
/// <param name="mode">The driver's ingest shape — chosen by driver config, never sniffed from the tag.</param>
/// <param name="def">The resolved tag definition.</param>
/// <param name="value">The value to write.</param>
/// <param name="encoded">The topic + payload to publish, when this returns <see cref="MqttWriteEncodeError.None"/>.</param>
/// <returns>The encode outcome.</returns>
/// <remarks>
/// <b>QoS 0, retain <see langword="false"/>.</b> 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 <see cref="RebirthRequester"/> avoids for NCMD, and the reason neither is
/// configurable.
/// </remarks>
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);
}
/// <summary>Encodes a Plain-mode write: the value's text/JSON rendering onto the tag's write topic.</summary>
/// <param name="def">The resolved tag definition.</param>
/// <param name="value">The value to write.</param>
/// <param name="encoded">The topic + payload on success.</param>
/// <returns>The encode outcome.</returns>
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;
}
/// <summary>Encodes a Sparkplug-mode write as an NCMD (node metric) or DCMD (device metric).</summary>
/// <param name="def">The resolved tag definition.</param>
/// <param name="value">The value to write.</param>
/// <param name="encoded">The topic + payload on success.</param>
/// <returns>The encode outcome.</returns>
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;
}
/// <summary>
/// Sets the metric's typed value field and its <c>Datatype</c> from the tag's declared type.
/// </summary>
/// <param name="metric">The metric being built.</param>
/// <param name="dataType">The tag's declared driver data type.</param>
/// <param name="value">The value to convert.</param>
/// <returns>The encode outcome.</returns>
/// <remarks>
/// <c>Datatype</c> is set explicitly rather than left for the node to infer, for the same reason
/// <see cref="RebirthRequester"/> 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.
/// </remarks>
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;
}
}
}
/// <summary>One encoded outbound write: where it goes and what it carries.</summary>
/// <param name="Topic">The concrete topic to publish on.</param>
/// <param name="Payload">The encoded message body.</param>
public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload);
@@ -0,0 +1,244 @@
using System.Net.Security;
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using Microsoft.Extensions.Logging;
using MQTTnet;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// <summary>
/// Assembles the MQTTnet client options for every broker connection this repo opens — the runtime
/// driver's, the browse session's and the Test-connect probe's alike.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why this is its own type in its own assembly (Gitea #512).</b> It owns the security posture of
/// an MQTT connection: whether TLS is on, whether an untrusted certificate is tolerated, and — when a
/// PEM CA is pinned — the custom chain validator that replaces the OS trust anchor. Both the runtime
/// driver and the browse session must connect with <em>identical</em> semantics, and the failure mode
/// of a second copy is not a crash but a silent divergence: browse quietly not honouring a pinned CA
/// while the driver does. Previously this lived on <c>MqttConnection</c> and <c>.Browser</c> took a
/// documented layering exception to reach it; the exception is gone, and this is the one copy.
/// </para>
/// <para>
/// <b>Pure.</b> No I/O, no network, nothing read from disk — so it is safe to call on a
/// config-validation path. A configured <see cref="MqttDriverOptions.CaCertificatePath"/> is read
/// lazily by the returned validation callback at handshake time, not here.
/// </para>
/// </remarks>
public static class MqttClientOptionsFactory
{
/// <summary>
/// Assembles the MQTTnet client options from <paramref name="options"/>. Pure — it performs
/// no I/O and touches no network, so it is safe to call on a config-validation path. A
/// configured <see cref="MqttDriverOptions.CaCertificatePath"/> is read lazily by the
/// returned certificate-validation callback at handshake time, not here.
/// </summary>
/// <param name="options">Broker connection settings.</param>
/// <param name="clientIdSuffix">
/// Appended to <see cref="MqttDriverOptions.ClientId"/>, so a browse/probe session can share
/// the configured identity without colliding with the driver's own client id. When both are
/// empty MQTTnet generates a random client id.
/// </param>
/// <param name="logger">Optional logger for the deferred CA-pin load; never receives credentials.</param>
public static MqttClientOptions BuildClientOptions(
MqttDriverOptions options,
string? clientIdSuffix,
ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
var builder = new MqttClientOptionsBuilder()
.WithTcpServer(options.Host, options.Port)
.WithProtocolVersion(MapProtocolVersion(options.ProtocolVersion))
.WithCleanSession(options.CleanSession)
.WithKeepAlivePeriod(TimeSpan.FromSeconds(options.KeepAliveSeconds))
.WithTimeout(TimeSpan.FromSeconds(options.ConnectTimeoutSeconds));
var clientId = string.Concat(options.ClientId ?? string.Empty, clientIdSuffix ?? string.Empty);
if (!string.IsNullOrWhiteSpace(clientId))
{
builder = builder.WithClientId(clientId);
}
// An empty username means "connect anonymously"; MQTTnet would otherwise send an empty
// CONNECT username, which some brokers treat as a failed auth rather than as anonymous.
if (!string.IsNullOrEmpty(options.Username))
{
builder = builder.WithCredentials(options.Username, options.Password ?? string.Empty);
}
builder = builder.WithTlsOptions(tls => ConfigureTls(tls, options, logger));
return builder.Build();
}
private static void ConfigureTls(MqttClientTlsOptionsBuilder tls, MqttDriverOptions options, ILogger? logger)
{
if (!options.UseTls)
{
tls.UseTls(false);
return;
}
tls.UseTls(true)
// Explicit SNI / hostname-verification target. Left unset, name validation depends on
// how MQTTnet infers the host from the endpoint — pin it to the configured host.
.WithTargetHost(options.Host);
if (options.AllowUntrustedServerCertificate)
{
// Dev / on-prem escape hatch, off unless explicitly enabled — mirrors the
// ServerHistorian TLS knobs. This is the ONLY branch that accepts a bad certificate.
tls.WithAllowUntrustedCertificates(true)
.WithIgnoreCertificateChainErrors(true)
.WithCertificateValidationHandler(static _ => true);
return;
}
if (string.IsNullOrWhiteSpace(options.CaCertificatePath))
{
// No pin: MQTTnet's default handler validates against the OS trust store.
return;
}
tls.WithCertificateValidationHandler(CreatePinnedCaValidator(options.CaCertificatePath, logger));
}
/// <summary>
/// Builds a certificate-validation callback that pins the broker chain to the PEM CA at
/// <paramref name="caCertificatePath"/>. The file is loaded lazily on first validation
/// (i.e. during the TLS handshake) so that this — and therefore
/// <see cref="BuildClientOptions"/> — stays free of I/O. Any failure to load or to chain
/// up to the pinned CA rejects the certificate: fail closed.
/// </summary>
private static Func<MqttClientCertificateValidationEventArgs, bool> CreatePinnedCaValidator(
string caCertificatePath,
ILogger? logger)
{
var trustedRoots = new Lazy<X509Certificate2Collection?>(
() => LoadPemCa(caCertificatePath, logger),
LazyThreadSafetyMode.ExecutionAndPublication);
return args => ValidateAgainstPinnedCa(args, trustedRoots.Value, caCertificatePath, logger);
}
private static X509Certificate2Collection? LoadPemCa(string caCertificatePath, ILogger? logger)
{
try
{
var collection = new X509Certificate2Collection();
collection.ImportFromPemFile(caCertificatePath);
if (collection.Count != 0)
{
return collection;
}
logger?.LogError("MQTT CA pin '{CaCertificatePath}' contains no certificates; broker TLS will be rejected.", caCertificatePath);
}
catch (Exception ex)
{
logger?.LogError(ex, "MQTT CA pin '{CaCertificatePath}' could not be loaded; broker TLS will be rejected.", caCertificatePath);
}
return null;
}
private static bool ValidateAgainstPinnedCa(
MqttClientCertificateValidationEventArgs args,
X509Certificate2Collection? trustedRoots,
string caCertificatePath,
ILogger? logger)
{
if (trustedRoots is null || trustedRoots.Count == 0)
{
// Unreadable pin ⇒ no trust anchor ⇒ refuse. Never silently degrade to the OS store.
return false;
}
// The pin replaces the trust anchor only. Anything else the platform flagged — a hostname
// mismatch, an absent certificate — remains fatal.
if ((args.SslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None)
{
logger?.LogError("MQTT broker certificate rejected: {SslPolicyErrors}.", args.SslPolicyErrors);
return false;
}
if (args.Certificate is null)
{
return false;
}
var presented = args.Certificate as X509Certificate2;
X509Certificate2? owned = null;
if (presented is null)
{
owned = X509CertificateLoader.LoadCertificate(args.Certificate.Export(X509ContentType.Cert));
presented = owned;
}
try
{
using var chain = new X509Chain();
chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust;
chain.ChainPolicy.CustomTrustStore.AddRange(trustedRoots);
// Revocation is deliberately not checked: the pin exists precisely for self-issued
// dev / on-prem CAs, which typically publish no CRL or OCSP responder, so checking
// would fail every such chain. The pin itself is the revocation story — remove the CA.
chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
// Seed the intermediates the broker presented during the handshake. A leaf signed by
// an intermediate that chains up to the pinned root is a common broker topology, and
// that intermediate arrives on the wire rather than being installed locally — without
// it in the candidate pool a legitimate trust path simply cannot be built.
// CustomRootTrust still means only `trustedRoots` may terminate the chain, so a rogue
// self-signed cert smuggled in here cannot become an anchor.
// Both sources are read: platforms differ in whether the peer-supplied intermediates
// land in the incoming chain's elements, its ExtraStore, or both.
if (args.Chain is not null)
{
foreach (var element in args.Chain.ChainElements)
{
chain.ChainPolicy.ExtraStore.Add(element.Certificate);
}
chain.ChainPolicy.ExtraStore.AddRange(args.Chain.ChainPolicy.ExtraStore);
}
if (chain.Build(presented))
{
return true;
}
logger?.LogError(
"MQTT broker certificate does not chain to the pinned CA '{CaCertificatePath}': {ChainStatus}.",
caCertificatePath,
string.Join(", ", chain.ChainStatus.Select(s => s.Status)));
return false;
}
catch (CryptographicException ex)
{
// X509Chain.Build throws on a malformed certificate or an unusable policy. An exception
// escaping a TLS validation callback surfaces as an opaque handshake crash, so classify
// it here and refuse: an unverifiable certificate is a rejected certificate.
logger?.LogError(
ex,
"MQTT broker certificate could not be validated against the pinned CA '{CaCertificatePath}'; rejecting.",
caCertificatePath);
return false;
}
finally
{
owned?.Dispose();
}
}
private static MQTTnet.Formatter.MqttProtocolVersion MapProtocolVersion(MqttProtocolVersion version)
=> version switch
{
MqttProtocolVersion.V311 => MQTTnet.Formatter.MqttProtocolVersion.V311,
MqttProtocolVersion.V500 => MQTTnet.Formatter.MqttProtocolVersion.V500,
_ => throw new ArgumentOutOfRangeException(nameof(version), version, "Unsupported MQTT protocol version."),
};
}
@@ -0,0 +1,40 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<LangVersion>latest</LangVersion>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<RootNamespace>ZB.MOM.WW.OtOpcUa.Driver.Mqtt</RootNamespace>
<AssemblyName>ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport</AssemblyName>
</PropertyGroup>
<!--
The leaf project the .Browser csproj's "KNOWN, DELIBERATE EXCEPTION" comment called for, now
that the exception is gone (Gitea #512).
It exists for exactly one reason: MqttClientOptionsFactory owns the TLS posture, the PEM CA-pin
chain validator, the credential handling and the protocol-version mapping — and BOTH the runtime
driver and the browse session have to open a broker connection with identical semantics. A second
copy would be a silent security divergence: browse quietly not honouring a pinned CA while the
driver does.
Why not .Contracts, the obvious home: .Contracts is deliberately transport-free (it carries the
Sparkplug wire schema + codec and has no MQTTnet reference), and BuildClientOptions returns
MqttClientOptions — an MQTTnet type. Putting it there would hand MQTTnet to every .Contracts
consumer, including ControlPlane, which has no business with a broker client.
Kept deliberately thin: this holds connection POLICY only, and nothing that connects. The live
client, the reconnect loop and the subscription bookkeeping stay in .Driver.
-->
<ItemGroup>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
</ItemGroup>
<ItemGroup>
<PackageReference Include="MQTTnet"/>
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions"/>
</ItemGroup>
</Project>
@@ -568,7 +568,7 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport,
MqttClientOptions clientOptions;
try
{
clientOptions = BuildClientOptions(_options, clientIdSuffix: null, _logger);
clientOptions = MqttClientOptionsFactory.BuildClientOptions(_options, clientIdSuffix: null, _logger);
}
catch (Exception ex)
{
@@ -1345,217 +1345,4 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport,
}
}
/// <summary>
/// Assembles the MQTTnet client options from <paramref name="options"/>. Pure — it performs
/// no I/O and touches no network, so it is safe to call on a config-validation path. A
/// configured <see cref="MqttDriverOptions.CaCertificatePath"/> is read lazily by the
/// returned certificate-validation callback at handshake time, not here.
/// </summary>
/// <param name="options">Broker connection settings.</param>
/// <param name="clientIdSuffix">
/// Appended to <see cref="MqttDriverOptions.ClientId"/>, so a browse/probe session can share
/// the configured identity without colliding with the driver's own client id. When both are
/// empty MQTTnet generates a random client id.
/// </param>
/// <param name="logger">Optional logger for the deferred CA-pin load; never receives credentials.</param>
public static MqttClientOptions BuildClientOptions(
MqttDriverOptions options,
string? clientIdSuffix,
ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
var builder = new MqttClientOptionsBuilder()
.WithTcpServer(options.Host, options.Port)
.WithProtocolVersion(MapProtocolVersion(options.ProtocolVersion))
.WithCleanSession(options.CleanSession)
.WithKeepAlivePeriod(TimeSpan.FromSeconds(options.KeepAliveSeconds))
.WithTimeout(TimeSpan.FromSeconds(options.ConnectTimeoutSeconds));
var clientId = string.Concat(options.ClientId ?? string.Empty, clientIdSuffix ?? string.Empty);
if (!string.IsNullOrWhiteSpace(clientId))
{
builder = builder.WithClientId(clientId);
}
// An empty username means "connect anonymously"; MQTTnet would otherwise send an empty
// CONNECT username, which some brokers treat as a failed auth rather than as anonymous.
if (!string.IsNullOrEmpty(options.Username))
{
builder = builder.WithCredentials(options.Username, options.Password ?? string.Empty);
}
builder = builder.WithTlsOptions(tls => ConfigureTls(tls, options, logger));
return builder.Build();
}
private static void ConfigureTls(MqttClientTlsOptionsBuilder tls, MqttDriverOptions options, ILogger? logger)
{
if (!options.UseTls)
{
tls.UseTls(false);
return;
}
tls.UseTls(true)
// Explicit SNI / hostname-verification target. Left unset, name validation depends on
// how MQTTnet infers the host from the endpoint — pin it to the configured host.
.WithTargetHost(options.Host);
if (options.AllowUntrustedServerCertificate)
{
// Dev / on-prem escape hatch, off unless explicitly enabled — mirrors the
// ServerHistorian TLS knobs. This is the ONLY branch that accepts a bad certificate.
tls.WithAllowUntrustedCertificates(true)
.WithIgnoreCertificateChainErrors(true)
.WithCertificateValidationHandler(static _ => true);
return;
}
if (string.IsNullOrWhiteSpace(options.CaCertificatePath))
{
// No pin: MQTTnet's default handler validates against the OS trust store.
return;
}
tls.WithCertificateValidationHandler(CreatePinnedCaValidator(options.CaCertificatePath, logger));
}
/// <summary>
/// Builds a certificate-validation callback that pins the broker chain to the PEM CA at
/// <paramref name="caCertificatePath"/>. The file is loaded lazily on first validation
/// (i.e. during the TLS handshake) so that this — and therefore
/// <see cref="BuildClientOptions"/> — stays free of I/O. Any failure to load or to chain
/// up to the pinned CA rejects the certificate: fail closed.
/// </summary>
private static Func<MqttClientCertificateValidationEventArgs, bool> CreatePinnedCaValidator(
string caCertificatePath,
ILogger? logger)
{
var trustedRoots = new Lazy<X509Certificate2Collection?>(
() => LoadPemCa(caCertificatePath, logger),
LazyThreadSafetyMode.ExecutionAndPublication);
return args => ValidateAgainstPinnedCa(args, trustedRoots.Value, caCertificatePath, logger);
}
private static X509Certificate2Collection? LoadPemCa(string caCertificatePath, ILogger? logger)
{
try
{
var collection = new X509Certificate2Collection();
collection.ImportFromPemFile(caCertificatePath);
if (collection.Count != 0)
{
return collection;
}
logger?.LogError("MQTT CA pin '{CaCertificatePath}' contains no certificates; broker TLS will be rejected.", caCertificatePath);
}
catch (Exception ex)
{
logger?.LogError(ex, "MQTT CA pin '{CaCertificatePath}' could not be loaded; broker TLS will be rejected.", caCertificatePath);
}
return null;
}
private static bool ValidateAgainstPinnedCa(
MqttClientCertificateValidationEventArgs args,
X509Certificate2Collection? trustedRoots,
string caCertificatePath,
ILogger? logger)
{
if (trustedRoots is null || trustedRoots.Count == 0)
{
// Unreadable pin ⇒ no trust anchor ⇒ refuse. Never silently degrade to the OS store.
return false;
}
// The pin replaces the trust anchor only. Anything else the platform flagged — a hostname
// mismatch, an absent certificate — remains fatal.
if ((args.SslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None)
{
logger?.LogError("MQTT broker certificate rejected: {SslPolicyErrors}.", args.SslPolicyErrors);
return false;
}
if (args.Certificate is null)
{
return false;
}
var presented = args.Certificate as X509Certificate2;
X509Certificate2? owned = null;
if (presented is null)
{
owned = X509CertificateLoader.LoadCertificate(args.Certificate.Export(X509ContentType.Cert));
presented = owned;
}
try
{
using var chain = new X509Chain();
chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust;
chain.ChainPolicy.CustomTrustStore.AddRange(trustedRoots);
// Revocation is deliberately not checked: the pin exists precisely for self-issued
// dev / on-prem CAs, which typically publish no CRL or OCSP responder, so checking
// would fail every such chain. The pin itself is the revocation story — remove the CA.
chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
// Seed the intermediates the broker presented during the handshake. A leaf signed by
// an intermediate that chains up to the pinned root is a common broker topology, and
// that intermediate arrives on the wire rather than being installed locally — without
// it in the candidate pool a legitimate trust path simply cannot be built.
// CustomRootTrust still means only `trustedRoots` may terminate the chain, so a rogue
// self-signed cert smuggled in here cannot become an anchor.
// Both sources are read: platforms differ in whether the peer-supplied intermediates
// land in the incoming chain's elements, its ExtraStore, or both.
if (args.Chain is not null)
{
foreach (var element in args.Chain.ChainElements)
{
chain.ChainPolicy.ExtraStore.Add(element.Certificate);
}
chain.ChainPolicy.ExtraStore.AddRange(args.Chain.ChainPolicy.ExtraStore);
}
if (chain.Build(presented))
{
return true;
}
logger?.LogError(
"MQTT broker certificate does not chain to the pinned CA '{CaCertificatePath}': {ChainStatus}.",
caCertificatePath,
string.Join(", ", chain.ChainStatus.Select(s => s.Status)));
return false;
}
catch (CryptographicException ex)
{
// X509Chain.Build throws on a malformed certificate or an unusable policy. An exception
// escaping a TLS validation callback surfaces as an opaque handshake crash, so classify
// it here and refuse: an unverifiable certificate is a rejected certificate.
logger?.LogError(
ex,
"MQTT broker certificate could not be validated against the pinned CA '{CaCertificatePath}'; rejecting.",
caCertificatePath);
return false;
}
finally
{
owned?.Dispose();
}
}
private static MQTTnet.Formatter.MqttProtocolVersion MapProtocolVersion(MqttProtocolVersion version)
=> version switch
{
MqttProtocolVersion.V311 => MQTTnet.Formatter.MqttProtocolVersion.V311,
MqttProtocolVersion.V500 => MQTTnet.Formatter.MqttProtocolVersion.V500,
_ => throw new ArgumentOutOfRangeException(nameof(version), version, "Unsupported MQTT protocol version."),
};
}
@@ -63,8 +63,31 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// </para>
/// </remarks>
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.
/// <summary>The reference is not an authored tag of this driver's live ingest path.</summary>
private const uint StatusBadNodeIdUnknown = 0x80340000u;
/// <summary>
/// The tag declares no write destination — a Plain tag with no <c>writeTopic</c>, 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.
/// </summary>
private const uint StatusBadNotWritable = 0x803B0000u;
/// <summary>The supplied value cannot be represented in the tag's declared data type.</summary>
private const uint StatusBadTypeMismatch = 0x80740000u;
/// <summary>The publish failed, or the driver has no live broker connection to publish through.</summary>
private const uint StatusBadCommunicationError = 0x80050000u;
/// <summary>
/// Rough per-tag driver-attributable cost: the parsed <see cref="MqttTagDefinition"/> (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<IReadOnlyList<DataValueSnapshot>>(results);
}
// ---- IWritable (Gitea #508) ----
/// <summary>
/// Whether <paramref name="def"/> declares somewhere for a write to go: a <c>writeTopic</c> in
/// Plain mode, or a complete command tuple in Sparkplug mode.
/// </summary>
/// <param name="def">The resolved tag definition.</param>
/// <returns><see langword="true"/> when a write to this tag can be addressed.</returns>
/// <remarks>
/// 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.
/// </remarks>
private bool CanWrite(MqttTagDefinition def)
=> MqttWriteEncoder.TryEncode(_options.Mode, def, ProbeValueFor(def.DataType), out _)
!= MqttWriteEncodeError.NoWriteDestination;
/// <summary>A never-published sentinel value of the given type, for the <see cref="CanWrite"/> probe.</summary>
/// <param name="dataType">The tag's declared data type.</param>
/// <returns>A representable value, or a string for types with no natural zero.</returns>
private static object ProbeValueFor(DriverDataType dataType) => dataType switch
{
DriverDataType.Boolean => false,
DriverDataType.DateTime => DateTime.UnixEpoch,
DriverDataType.String or DriverDataType.Reference => "",
_ => 0,
};
/// <summary>
/// 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).
/// </summary>
/// <param name="writes">The requested writes.</param>
/// <param name="cancellationToken">Cancellation for the batch.</param>
/// <returns>One result per request, in request order.</returns>
/// <remarks>
/// <para>
/// <b>Never throws, and one bad request never fails the batch.</b> Each slot carries its own
/// status, matching <see cref="ReadAsync"/>: failing the whole call would let a single stale
/// reference reject every other write in the same OPC UA request.
/// </para>
/// <para>
/// <b>Good means "published", not "applied".</b> 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.
/// </para>
/// <para>
/// <b>Consequence, inherited knowingly (see #5, write-outcome self-correction):</b> because the
/// transport cannot report device-level rejection, this driver can never surface a failed
/// <em>device</em> 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.
/// </para>
/// </remarks>
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
IReadOnlyList<WriteRequest> 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;
}
/// <summary>Maps an encode refusal to the status an OPC UA client sees.</summary>
/// <param name="error">The encode outcome.</param>
/// <returns>The OPC UA status code.</returns>
private static uint MapEncodeError(MqttWriteEncodeError error) => error switch
{
MqttWriteEncodeError.NoWriteDestination => StatusBadNotWritable,
MqttWriteEncodeError.UnsupportedDataType => StatusBadNotWritable,
_ => StatusBadTypeMismatch,
};
// ---- IHostConnectivityProbe ----
/// <summary>The single broker this driver instance talks to, as <c>host:port</c>.</summary>
@@ -939,7 +1103,7 @@ public sealed class MqttDriver
// ONLINE half without the Will is strictly worse than shipping neither: a crashed node
// would leave a retained ONLINE asserting forever that a dead host is alive, which is the
// exact state the mechanism exists to prevent. The Will belongs in
// MqttConnection.BuildClientOptions; until it is there this flag stays inert and said so.
// MqttClientOptionsFactory.BuildClientOptions; until it is there this flag stays inert and said so.
_logger?.LogWarning(
"MQTT driver '{DriverId}': Sparkplug actAsPrimaryHost is set but primary-host STATE "
+ "publishing is not implemented — this driver observes STATE and never publishes one. "
@@ -17,7 +17,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// </summary>
/// <remarks>
/// <para>
/// <b>Reuses <see cref="MqttConnection.BuildClientOptions"/></b> for the connect (TLS,
/// <b>Reuses <see cref="MqttClientOptionsFactory.BuildClientOptions"/></b> for the connect (TLS,
/// CA-pin, credentials, protocol-version mapping) rather than hand-rolling a second
/// connect path — a duplicate would be a security divergence. A distinct client-id suffix
/// (see <see cref="ProbeClientIdPrefix"/>) keeps a probe from colliding with the driver's
@@ -101,7 +101,7 @@ public sealed class MqttDriverProbe : IDriverProbe
MqttClientOptions clientOptions;
try
{
clientOptions = MqttConnection.BuildClientOptions(
clientOptions = MqttClientOptionsFactory.BuildClientOptions(
probeOptions,
ProbeClientIdPrefix + Guid.NewGuid().ToString("N")[..8],
NullLogger.Instance);
@@ -1038,7 +1038,7 @@ public sealed class SparkplugIngestor
/// while their host is offline. Shipping the ONLINE half without the Will is strictly worse
/// than shipping neither: an OtOpcUa node that crashes would leave a retained ONLINE asserting
/// forever that a dead host is alive, which is precisely the state the mechanism exists to
/// prevent. The Will has to be set inside <c>MqttConnection.BuildClientOptions</c> — a
/// prevent. The Will has to be set inside <c>MqttClientOptionsFactory.BuildClientOptions</c> — a
/// different file, a different task — so this task implements the observation half and
/// <see cref="MqttDriver"/> warns when the flag is set. See the design doc §7/§8.
/// </para>
@@ -14,6 +14,7 @@
<ItemGroup>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport.csproj"/>
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core\ZB.MOM.WW.OtOpcUa.Core.csproj"/>
</ItemGroup>
@@ -0,0 +1,80 @@
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Browsing;
/// <summary>
/// Validates an operator-typed Sparkplug rebirth scope for the raw browse modal (Gitea #514).
/// </summary>
/// <remarks>
/// <para>
/// <b>Why a typed scope exists at all.</b> The Sparkplug browse tree is <i>birth-driven</i>, and
/// births are never retained. On a quiet plant — every edge node birthed long before the browse
/// window opened, and is now publishing only NDATA/DDATA — the tree comes up empty. The
/// Request-rebirth scope was derived solely from a tree selection, so the one action that would
/// populate the tree could not be armed: the action needed the tree, and the tree needed the action.
/// </para>
/// <para>
/// <b>This is advisory, not the gate.</b> <c>MqttBrowseSession.ResolveRebirthTargets</c> re-resolves
/// the scope and refuses an unusable one <em>before any message leaves</em>, and the
/// <c>DriverOperator</c> policy is enforced server-side in <c>BrowserSessionService</c>. This exists
/// so the modal does not let an operator arm a two-click confirm that cannot succeed. Being the
/// looser of the two is therefore safe; being the stricter one would block a scope the session
/// accepts, which is the failure mode the repo's authoring surfaces avoid by convention.
/// </para>
/// <para>
/// <b>A group-only scope is rejected on purpose.</b> The session resolves a bare group by enumerating
/// the edge nodes it has <i>observed</i> — of which an empty tree has none — so it could not help the
/// case this input exists for, and on a populated tree it would fan one click across the whole group.
/// The tree-selection path still offers a group scope, with its own confirm-time cap and warning.
/// </para>
/// </remarks>
public static class SparkplugRebirthScope
{
/// <summary>MQTT wildcards. A rebirth addresses one edge node exactly; a wildcard cannot name one.</summary>
private static readonly char[] Wildcards = ['+', '#'];
/// <summary>
/// Describes why <paramref name="raw"/> is not a usable <c>{group}/{edgeNode}</c> rebirth scope.
/// </summary>
/// <param name="raw">The operator-typed scope, untrimmed; <see langword="null"/> is treated as blank.</param>
/// <returns>The operator-facing error text, or <see langword="null"/> when the scope is usable.</returns>
public static string? Describe(string? raw)
{
var scope = (raw ?? "").Trim();
if (scope.Length == 0) { return "Enter the edge node as 'group/edgeNode'."; }
if (scope.IndexOfAny(Wildcards) >= 0)
{
return "An MQTT wildcard (+ or #) cannot address an edge node; name one exactly.";
}
var segments = scope.Split('/');
if (segments.Length != 2 || Array.Exists(segments, string.IsNullOrWhiteSpace))
{
return "Use exactly 'group/edgeNode' — a group on its own would ask every edge node in it to "
+ "republish, which this window cannot enumerate while the tree is empty.";
}
return null;
}
/// <summary>True when <paramref name="raw"/> is a usable rebirth scope.</summary>
/// <param name="raw">The operator-typed scope.</param>
/// <returns><see langword="true"/> when <see cref="Describe"/> finds no fault.</returns>
public static bool IsUsable(string? raw) => Describe(raw) is null;
/// <summary>
/// Splits a scope already accepted by <see cref="Describe"/> into its two segments.
/// </summary>
/// <param name="raw">A scope for which <see cref="IsUsable"/> is <see langword="true"/>.</param>
/// <returns>The trimmed <c>(group, edgeNode)</c> pair.</returns>
/// <exception cref="ArgumentException">The scope is not usable.</exception>
public static (string GroupId, string EdgeNodeId) Split(string? raw)
{
if (Describe(raw) is { } error)
{
throw new ArgumentException($"'{raw}' is not a usable rebirth scope: {error}", nameof(raw));
}
var segments = raw!.Trim().Split('/');
return (segments[0].Trim(), segments[1].Trim());
}
}
@@ -126,9 +126,34 @@
@if (_rebirthTarget is null)
{
@* Gitea #514: the tree is BIRTH-driven, and births are never retained. On a
quiet plant — every edge node birthed long before this window opened and is
now publishing only NDATA/DDATA — the tree is empty, so a tree-selected
scope can never be armed and the one action that would fill the tree is
unreachable. Naming an edge node directly breaks that cycle. Deliberately
an EDGE NODE rather than a group broadcast: a group rebirth is a
self-inflicted broker storm on a large plant. *@
<div class="small text-muted mt-1">
Click a group, edge node, device or metric in the tree above to choose the scope.
Click a group, edge node, device or metric in the tree above to choose the
scope &mdash; or, if the tree is empty because every edge node birthed
before this window opened, name one directly:
</div>
<div class="input-group input-group-sm mt-1">
<span class="input-group-text">Edge node</span>
<input class="form-control mono" placeholder="group/edgeNode"
aria-label="Sparkplug edge-node scope"
value="@_manualScope"
@oninput="@(e => _manualScope = e.Value?.ToString() ?? "")" />
<button type="button" class="btn btn-outline-secondary"
disabled="@(ManualScopeError is not null)"
@onclick="ApplyManualScope">
Use this scope
</button>
</div>
@if (!string.IsNullOrWhiteSpace(_manualScope) && ManualScopeError is { } manualErr)
{
<div class="small text-danger mt-1">@manualErr</div>
}
}
else
{
@@ -179,6 +204,17 @@
}
</div>
}
else if (_rebirthExpired && _canOperate)
{
@* Gitea #510: the affordance is retired rather than left to fail on click. It is
drawn from a capability probed at browse-open, and sessions are reaped for
idleness — so a modal left open past the reap used to offer an action whose
token no longer resolved. Say what happened instead of silently vanishing. *@
<div class="alert alert-warning py-2 mt-2 mb-0 small">
This browse session expired while idle, so a rebirth can no longer be requested
through it. Close and re-open the browser to request one.
</div>
}
<div class="mt-2 small">
<strong>@_selected.Count</strong> tag@(_selected.Count == 1 ? "" : "s") selected.
@@ -276,6 +312,18 @@
private string? _rebirthOutcome;
private string? _rebirthError;
/// <summary>
/// Set when a re-probe finds the session's rebirth capability gone (#510), so the retired
/// affordance can be explained rather than silently disappearing.
/// </summary>
private bool _rebirthExpired;
/// <summary>
/// Operator-typed <c>{group}/{edgeNode}</c> rebirth scope, for the birth-driven-tree deadlock
/// (#514). Seeded with the driver's configured group so only the edge node has to be typed.
/// </summary>
private string _manualScope = "";
private readonly Dictionary<string, SelectedLeaf> _selected = new(StringComparer.Ordinal);
private readonly HashSet<string> _selectedIds = new(StringComparer.Ordinal);
@@ -316,6 +364,8 @@
_disabledReason = null;
ResetRebirthState();
_canRebirth = false;
_rebirthExpired = false;
_manualScope = "";
_resolving = true;
StateHasChanged();
@@ -373,8 +423,16 @@
{
_token = result.Token;
// Capability, not authorization: a plain-MQTT window (and every non-MQTT browser) has no
// re-announce action at all, so the affordance must not be drawn for it.
// re-announce action at all, so the affordance must not be drawn for it. Re-probed per
// click as well — see RebirthStillAvailable (#510).
_canRebirth = BrowserService.CanRequestRebirth(_token);
_rebirthExpired = false;
// Only the edge node should have to be typed (#514): the group is the driver's entire
// subscription scope, so it is already in the config this browser was opened with.
if (_canRebirth && _manualScope.Length == 0 && TryReadConfiguredGroupId() is { } gid)
{
_manualScope = gid + "/";
}
}
else { _openError = result.Message; }
}
@@ -502,9 +560,64 @@
{
_rebirthOutcome = null;
_rebirthError = null;
// #510 — the capability was probed once at browse-open; re-probe before offering to act on it.
if (!RebirthStillAvailable()) { return; }
_rebirthConfirming = true;
}
/// <summary>
/// Re-probes the open session's rebirth capability, retiring the affordance when it has gone
/// (Gitea #510).
/// </summary>
/// <returns><see langword="true"/> when a rebirth can still be requested through this session.</returns>
/// <remarks>
/// <c>CanRequestRebirth</c> is a cheap, non-throwing, no-I/O probe that is false for a reaped or
/// unknown token, so re-running it per click costs nothing. It is a <b>capability</b> probe, not
/// an authorization check — the <c>DriverOperator</c> gate stays server-side on
/// <c>RequestRebirthAsync</c> and fails closed there, and <c>_canOperate</c> remains the
/// defence-in-depth half of the render condition.
/// </remarks>
private bool RebirthStillAvailable()
{
if (BrowserService.CanRequestRebirth(_token)) { return true; }
_canRebirth = false;
_rebirthExpired = true;
ResetRebirthState();
return false;
}
/// <summary>
/// Describes why <see cref="_manualScope"/> is not a usable edge-node scope, or <c>null</c> when
/// it is. Mirrors the session's own <c>{group}/{edgeNode}</c> rule (Gitea #514).
/// </summary>
/// <remarks>
/// Advisory only: <c>MqttBrowseSession.ResolveRebirthTargets</c> re-validates and refuses an
/// unresolvable scope <b>before publishing anything</b>, so this exists to keep the operator from
/// arming a confirm step that cannot succeed — not as the gate. A group-only scope is rejected on
/// purpose: the session would fan out to every observed edge node, and on a tree that is empty
/// (the case this input exists for) there are none to fan to.
/// </remarks>
private string? ManualScopeError => SparkplugRebirthScope.Describe(_manualScope);
/// <summary>Adopts the typed <c>{group}/{edgeNode}</c> as the rebirth scope (Gitea #514).</summary>
private void ApplyManualScope()
{
if (!RebirthStillAvailable()) { return; }
if (!SparkplugRebirthScope.IsUsable(_manualScope)) { return; }
var (groupId, edgeNodeId) = SparkplugRebirthScope.Split(_manualScope);
_rebirthTarget = new RebirthTarget(
$"{groupId}/{edgeNodeId}",
$"edge node {edgeNodeId}",
$"Edge node '{edgeNodeId}' in group '{groupId}' republishes its whole metric set. "
+ "This node was named directly, so it need not have appeared in the tree.",
IsGroup: false);
_rebirthConfirming = false;
_rebirthOutcome = null;
_rebirthError = null;
}
/// <summary>Disarms the confirm step without publishing anything.</summary>
private void CancelRebirth() => _rebirthConfirming = false;
@@ -516,6 +629,9 @@
private async Task ConfirmRebirthAsync()
{
if (_rebirthTarget is null) { return; }
// #510 again, at the last moment before publishing: the reap can land between arming the confirm
// step and clicking it, and this is the one path that reaches a live broker.
if (!RebirthStillAvailable()) { return; }
_rebirthBusy = true;
_rebirthOutcome = null;
_rebirthError = null;
@@ -558,6 +674,34 @@
ResetRebirthState();
}
/// <summary>
/// Reads the driver's configured Sparkplug group id out of the merged config, or <c>null</c> when
/// it is absent or the blob is unreadable. Purely a convenience seed for <see cref="_manualScope"/>
/// — never a gate, so any failure is simply "no seed".
/// </summary>
/// <returns>The configured group id, or <c>null</c>.</returns>
private string? TryReadConfiguredGroupId()
{
try
{
// Case-insensitive: the merged blob is assembled from the driver + device configs and this
// is a display convenience, not a binding — it must not depend on which one won.
var root = System.Text.Json.Nodes.JsonNode.Parse(_mergedConfig)?.AsObject();
var sparkplug = root?
.FirstOrDefault(p => string.Equals(p.Key, "Sparkplug", StringComparison.OrdinalIgnoreCase))
.Value?.AsObject();
var groupId = sparkplug?
.FirstOrDefault(p => string.Equals(p.Key, "GroupId", StringComparison.OrdinalIgnoreCase))
.Value?.GetValue<string>();
return string.IsNullOrWhiteSpace(groupId) ? null : groupId.Trim();
}
catch
{
return null;
}
}
/// <summary>Clears every rebirth-panel field (modal re-open, browser close).</summary>
private void ResetRebirthState()
{
@@ -586,12 +730,22 @@
.ToList();
if (_commitErrors.Count > 0) { return; }
var rows = _selected.Values
var selected = _selected.Values.ToList();
var rows = selected
.Select(s => RawBrowseCommitMapper.MapLeaf(
_driverType, s.NodeId, s.Name, s.DriverDataType, DefaultDataType,
_groupPrefix, s.FolderPath, _createGroups, s.AddressFields))
.ToList();
// A leaf name is coerced into a legal RawPath segment (#509), and that coercion is not
// injective — 'A/B' and 'A_B' both land on 'A_B'. Refuse the colliding SET by name here:
// letting it through would silently merge two distinct metrics into one tag bound to
// whichever address survived, which is worse than the outright refusal #509 replaced.
_commitErrors = RawBrowseCommitMapper
.DescribeNameCollisions(selected.Select(s => s.Name).ToList(), rows)
.ToList();
if (_commitErrors.Count > 0) { return; }
var outcome = await Svc.ImportTagsAsync(_effectiveDeviceId, rows);
if (outcome.Errors.Count > 0)
{
@@ -613,6 +767,8 @@
var t = _token;
_token = Guid.Empty;
_canRebirth = false;
_rebirthExpired = false;
_manualScope = "";
ResetRebirthState();
Visible = false;
_open = false;
@@ -30,6 +30,18 @@
@onchange="@(e => Update(() => _m.Topic = e.Value?.ToString() ?? ""))" />
<div class="form-text">Concrete topic — MQTT wildcards (+ / #) are not valid for a tag.</div>
</div>
<div class="col-md-8">
<label class="form-label" for="mqtt-write-topic">Write topic <span class="text-muted">(optional)</span></label>
<input id="mqtt-write-topic" type="text" class="form-control form-control-sm mono"
placeholder="factory/oven/setpoint/set" value="@_m.WriteTopic"
@onchange="@(e => Update(() => _m.WriteTopic = e.Value?.ToString() ?? ""))" />
<div class="form-text">
Where a write is published. Deliberately separate from the read topic — a tag commonly
reads <code>&hellip;/state</code> and writes <code>&hellip;/set</code>. Leave blank for a
read-only tag: a write is then refused (<code>BadNotWritable</code>) rather than published
nowhere.
</div>
</div>
<div class="col-md-4">
<label class="form-label" for="mqtt-payload-format">Payload format</label>
<select id="mqtt-payload-format" class="form-select form-select-sm" value="@_m.PayloadFormat"
@@ -25,7 +25,10 @@ public static class RawBrowseCommitMapper
/// <param name="driverType">The owning device's driver type (a <see cref="DriverTypeNames"/> value).</param>
/// <param name="fullName">The leaf's driver-side full reference (the <c>BrowseNode.NodeId</c> = the
/// captured <c>DriverAttributeInfo.FullName</c>) — becomes the driver-typed address field.</param>
/// <param name="browseName">The leaf's browse name — becomes the raw tag <c>Name</c> (a RawPath segment).</param>
/// <param name="browseName">The leaf's browse name. Becomes the raw tag <c>Name</c> after
/// <see cref="ToRawSegment"/> coerces it into a legal RawPath segment — a Sparkplug metric name
/// legitimately contains <c>/</c> and a Name may not (Gitea #509). The leaf's binding address is
/// unaffected: it travels separately into the <c>TagConfig</c>.</param>
/// <param name="driverDataType">The leaf's <see cref="DriverDataType"/> name (from the browse
/// attribute side-channel), or null when the browser cannot report a type (e.g. the OPC UA Client tree).</param>
/// <param name="defaultDataType">The OPC-UA built-in type name to fall back to when
@@ -52,6 +55,7 @@ public static class RawBrowseCommitMapper
{
var dataType = MapDataType(driverDataType) ?? defaultDataType;
var tagConfig = BuildTagConfig(driverType, fullName, addressFields);
var name = ToRawSegment(browseName);
var mirrored = createGroups && folderPath is { Count: > 0 }
? string.Join(RawPaths.Separator, folderPath)
: null;
@@ -61,9 +65,99 @@ public static class RawBrowseCommitMapper
// no poll-group (batching is authored later).
return new RawTagImportRow(
groupPath,
new RawTagInput(browseName, dataType, TagAccessLevel.Read, WriteIdempotent: false, PollGroupId: null, tagConfig));
new RawTagInput(name, dataType, TagAccessLevel.Read, WriteIdempotent: false, PollGroupId: null, tagConfig));
}
/// <summary>
/// Coerces a browse leaf's name into a legal RawPath segment (Gitea #509): trims the ends and
/// replaces every <see cref="RawPaths.Separator"/> with <see cref="SeparatorReplacement"/>.
/// </summary>
/// <param name="browseName">The leaf's browse name, as the driver reported it.</param>
/// <returns>A <see cref="RawPaths.ValidateSegment"/>-clean segment; blank only if the input was blank.</returns>
/// <remarks>
/// <para>
/// <b>The Name and the address are two fields, and only the Name is constrained.</b> A tag Name is
/// a RawPath segment, so it may not contain <c>/</c>; the driver-side reference it binds by has no
/// such rule and is written into the <c>TagConfig</c> untouched by <see cref="BuildTagConfig"/>.
/// Refusing the commit outright conflated the two — and it refused a metric the Sparkplug B spec
/// makes <em>mandatory</em> on every edge node (<c>Node Control/Rebirth</c>), so the one metric an
/// operator is most likely to browse for was the one that could not be committed.
/// </para>
/// <para>
/// <b>This deliberately covers every driver, not just MQTT.</b> The argument is about what a Name
/// is, not about who produced it: any browser whose leaf name carries a <c>/</c> hits the same
/// wall, and for all of them the binding address travels separately and survives intact.
/// </para>
/// <para>
/// <b>The rule set is <see cref="RawPaths.ValidateSegment"/>'s, not an invented one</b> — empty,
/// an embedded separator, and leading/trailing whitespace are exactly what it rejects, so trimming
/// and replacing is total rather than a best-effort guess. <c>RawBrowseCommitMapperTests</c> pins
/// that by asserting the output validates, so a new rule there fails here instead of silently
/// letting an illegal Name through to the import.
/// </para>
/// <para>
/// Sanitising is not injective, so two distinct leaves can land on one Name — see
/// <see cref="DescribeNameCollisions"/>, which is why a caller must run that check rather than
/// trusting this in isolation.
/// </para>
/// </remarks>
public static string ToRawSegment(string? browseName)
{
var trimmed = (browseName ?? "").Trim();
return trimmed.Length == 0 ? "" : trimmed.Replace(RawPaths.Separator, SeparatorReplacement);
}
/// <summary>What <see cref="ToRawSegment"/> substitutes for a separator inside a browse name.</summary>
private const char SeparatorReplacement = '_';
/// <summary>
/// Names the browse leaves that <see cref="ToRawSegment"/> would land on one tag Name within one
/// target group, or an empty list when the mapped rows are all distinct.
/// </summary>
/// <param name="browseNames">The ORIGINAL leaf names, in the same order as <paramref name="rows"/>.</param>
/// <param name="rows">The mapped import rows.</param>
/// <returns>One operator-facing message per colliding set; empty when there is nothing to refuse.</returns>
/// <remarks>
/// Sanitising maps many names onto one — <c>A/B</c> and <c>A_B</c> both become <c>A_B</c> — and a
/// silent collision is materially worse than the refusal #509 replaced: two distinct metrics would
/// become one tag bound to whichever address the import happened to keep. So the coercion is only
/// safe when paired with this check.
/// <para>
/// Scoped per target group, because the Name only has to be unique within its group — the same
/// metric name under two devices is not a collision.
/// </para>
/// <para>
/// The message names the specific pair rather than failing the batch anonymously. The commit
/// itself stays all-or-nothing, matching the CSV import's staged-review semantics: nothing is
/// written while any row is unresolvable.
/// </para>
/// </remarks>
public static IReadOnlyList<string> DescribeNameCollisions(
IReadOnlyList<string> browseNames,
IReadOnlyList<RawTagImportRow> rows)
{
ArgumentNullException.ThrowIfNull(browseNames);
ArgumentNullException.ThrowIfNull(rows);
return rows
.Select((row, i) => (row, original: i < browseNames.Count ? browseNames[i] : row.Tag.Name))
.GroupBy(x => (x.row.TagGroupPath ?? "", x.row.Tag.Name), TupleComparer)
.Where(g => g.Count() > 1)
.Select(g =>
$"'{g.Key.Item2}' would be the name of {g.Count()} different tags"
+ (string.IsNullOrEmpty(g.Key.Item1) ? "" : $" in group '{g.Key.Item1}'")
+ $": {string.Join(", ", g.Select(x => $"'{x.original}'"))}. "
+ "Their names differ only by characters that are not legal in a tag name. Commit them "
+ "separately, or author them by manual entry with distinct names.")
.ToList();
}
/// <summary>Ordinal comparer for the (group path, name) collision key — RawPath comparison is ordinal.</summary>
private static readonly IEqualityComparer<(string, string)> TupleComparer =
EqualityComparer<(string, string)>.Create(
(a, b) => RawPaths.Comparer.Equals(a.Item1, b.Item1) && RawPaths.Comparer.Equals(a.Item2, b.Item2),
x => HashCode.Combine(RawPaths.Comparer.GetHashCode(x.Item1), RawPaths.Comparer.GetHashCode(x.Item2)));
/// <summary>
/// Maps a <see cref="DriverDataType"/> enum NAME (as reported by the browse attribute side-channel) to the
/// OPC-UA built-in type name a <see cref="RawTagInput.DataType"/> carries. Mirrors
@@ -34,6 +34,7 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Uns.TagEditors;
public sealed class MqttTagConfigModel
{
private const string TopicKey = "topic";
private const string WriteTopicKey = "writeTopic";
private const string PayloadFormatKey = "payloadFormat";
private const string JsonPathKey = "jsonPath";
private const string DataTypeKey = "dataType";
@@ -80,6 +81,15 @@ public sealed class MqttTagConfigModel
/// <summary>The concrete MQTT topic the tag subscribes to (Plain mode). Required; no wildcards.</summary>
public string Topic { get; set; } = "";
/// <summary>
/// The topic a write PUBLISHES to (Plain mode), or blank when the tag is read-only (Gitea #508).
/// Deliberately distinct from <see cref="Topic"/> — a tag legitimately reads <c>.../state</c> and
/// writes <c>.../set</c> — and blank is meaningful: the driver refuses the write with
/// <c>BadNotWritable</c> rather than publishing nowhere. Unused in Sparkplug mode, where a
/// command is addressed by the binding tuple (NCMD/DCMD), not by a typed topic.
/// </summary>
public string WriteTopic { get; set; } = "";
/// <summary>How the received payload is decoded (Plain mode).</summary>
public MqttPayloadFormat PayloadFormat { get; set; } = MqttPayloadFormat.Json;
@@ -170,6 +180,7 @@ public sealed class MqttTagConfigModel
{
Mode = mode,
Topic = topic,
WriteTopic = TagConfigJson.GetString(o, WriteTopicKey) ?? "",
PayloadFormat = TagConfigJson.GetEnum(o, PayloadFormatKey, MqttPayloadFormat.Json),
JsonPath = jsonPath,
DataType = TagConfigJson.GetEnum(o, DataTypeKey, DriverDataType.String),
@@ -196,10 +207,12 @@ public sealed class MqttTagConfigModel
/// <c>dataType</c> is the one key both modes write, from two different typed fields
/// (<see cref="DataType"/> for Plain — always present; <see cref="MetricDataType"/> for
/// Sparkplug — omitted unless the operator set an override): which field wins is decided by
/// <see cref="Mode"/> at write time. Every other key is unconditional — mode-inapplicable keys
/// (e.g. <c>topic</c> while Sparkplug, or <c>groupId</c> while Plain) are written/cleared exactly
/// as their typed field says, which naturally preserves a retyped tag's other-mode leftovers
/// untouched until the operator edits that field too.
/// <see cref="Mode"/> at write time. <c>payloadFormat</c> is the one key written for <b>Plain
/// only</b> (Gitea #513) — see its line below for why it alone needs the mode test. Every other
/// key is unconditional: mode-inapplicable keys (e.g. <c>topic</c> while Sparkplug, or
/// <c>groupId</c> while Plain) are written/cleared exactly as their typed field says, and because
/// each of those fields is blank-⇒-omit, an absent key stays absent while a retyped tag's real
/// other-mode leftovers survive untouched until the operator edits that field too.
/// </remarks>
/// <returns>The serialised TagConfig JSON string.</returns>
public string ToJson()
@@ -207,7 +220,19 @@ public sealed class MqttTagConfigModel
// Blank ⇒ omit, like every other optional here: toggling the shape dropdown on an untouched tag
// must not leave a stray "topic":"" behind.
TagConfigJson.Set(_bag, TopicKey, string.IsNullOrWhiteSpace(Topic) ? null : Topic.Trim());
TagConfigJson.Set(_bag, PayloadFormatKey, PayloadFormat);
// Blank ⇒ omit ⇒ the tag is not writable. Written in BOTH modes, unlike payloadFormat: a
// Sparkplug tag ignores it, but silently deleting an operator's typed topic when they flip the
// shape dropdown would lose authored configuration rather than merely tidy the blob.
TagConfigJson.Set(_bag, WriteTopicKey, string.IsNullOrWhiteSpace(WriteTopic) ? null : WriteTopic.Trim());
// Plain-only (Gitea #513). Unlike every other Plain key here, PayloadFormat is a NON-NULLABLE
// typed field, so an unconditional write does not merely preserve a retyped tag's leftover — it
// INVENTS "payloadFormat":"Json" in the blob of a Sparkplug tag that never had one. The key names
// a decode path that Sparkplug does not use (MqttTagDefinitionFactory routes on the descriptor
// keys and never reads it), so it is a trap for anyone diagnosing a decode problem from the blob.
TagConfigJson.Set(_bag, PayloadFormatKey, Mode == MqttMode.SparkplugB ? null : (object)PayloadFormat);
TagConfigJson.Set(_bag, JsonPathKey, string.IsNullOrWhiteSpace(JsonPath) ? null : JsonPath.Trim());
TagConfigJson.Set(_bag, QosKey, Qos);
TagConfigJson.Set(_bag, RetainSeedKey, RetainSeed);
@@ -285,6 +310,15 @@ public sealed class MqttTagConfigModel
+ "or the tag would be fed by every matching topic.";
}
// A write topic is OPTIONAL (blank = read-only), but a present one must be publishable: a
// wildcard cannot be published to at all, so it would fail at the broker rather than here.
var writeTopic = WriteTopic.Trim();
if (writeTopic.Length > 0 && writeTopic.IndexOfAny(TopicWildcards) >= 0)
{
return $"Write topic '{writeTopic}' contains an MQTT wildcard (+ or #); a write is published "
+ "to one concrete topic.";
}
// NB no jsonPath rule — see the remarks above. A blank path is the driver's document-root
// default, not an authoring error.
return null;
@@ -0,0 +1,387 @@
using System.Buffers;
using System.Collections.Concurrent;
using System.Text;
using MQTTnet;
using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;
/// <summary>
/// Live validation of MQTT write-through (Gitea #508) against the real Mosquitto fixture: a Plain
/// publish to the tag's write topic, and Sparkplug NCMD/DCMD commands, observed on the broker by an
/// independent subscriber.
/// </summary>
/// <remarks>
/// <para>
/// <b>Observed on the broker, not on a fake.</b> The whole risk this feature carries is a write that
/// reports Good and reaches nothing, so the assertion has to be that a real message arrived on a
/// real broker at the topic the operator authored. A test double asserting "publish was called"
/// could not distinguish the feature working from the feature publishing to the wrong topic.
/// </para>
/// <para>
/// <b>Env-gated + skip-clean</b>, like every suite here: without <c>MQTT_FIXTURE_ENDPOINT</c> and
/// credentials each test calls <c>Assert.Skip</c>. See <see cref="MqttFixture"/>.
/// </para>
/// </remarks>
[Collection(MqttFixtureCollection.Name)]
[Trait("Category", "LiveIntegration")]
public sealed class MqttWriteLiveTests(MqttFixture fixture)
{
private const uint Good = 0x00000000u;
private const uint BadNotWritable = 0x803B0000u;
private static readonly TimeSpan MessageTimeout = TimeSpan.FromSeconds(30);
private readonly MqttFixture _fx = fixture;
private void SkipIfUnconfigured() => Assert.SkipWhen(_fx.SkipReason is not null, _fx.SkipReason ?? "");
// -----------------------------------------------------------------------
// Plain
// -----------------------------------------------------------------------
[Fact]
public async Task Plain_write_publishes_the_value_on_the_authored_write_topic()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
const string rawPath = "Fixture/Mqtt/Line1/Setpoint";
var readTopic = _fx.Topic($"write/{Guid.NewGuid():N}/state");
var writeTopic = _fx.Topic($"write/{Guid.NewGuid():N}/set");
await using var watcher = await TopicWatcher.StartAsync(_fx, writeTopic, ct);
var tag = new RawTagEntry(
rawPath,
$$"""{"topic":"{{readTopic}}","writeTopic":"{{writeTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
WriteIdempotent: false);
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write");
await driver.InitializeAsync(string.Empty, ct);
var results = await driver.WriteAsync([new WriteRequest(rawPath, 42.5)], ct);
results.ShouldHaveSingleItem().StatusCode.ShouldBe(Good);
var payload = await watcher.WaitForOneAsync(MessageTimeout, ct);
Encoding.UTF8.GetString(payload).ShouldBe(
"42.5",
"the write must arrive on the authored write topic carrying the value, in invariant form");
}
[Fact]
public async Task Plain_write_never_lands_on_the_tags_read_topic()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
const string rawPath = "Fixture/Mqtt/Line1/SetpointSplit";
var suffix = Guid.NewGuid().ToString("N");
var readTopic = _fx.Topic($"write/{suffix}/state");
var writeTopic = _fx.Topic($"write/{suffix}/set");
// Both watchers are live BEFORE the write, so "nothing on the read topic" is a real observation
// over the same window rather than an absence that was never looked for.
await using var readWatcher = await TopicWatcher.StartAsync(_fx, readTopic, ct);
await using var writeWatcher = await TopicWatcher.StartAsync(_fx, writeTopic, ct);
var tag = new RawTagEntry(
rawPath,
$$"""{"topic":"{{readTopic}}","writeTopic":"{{writeTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
WriteIdempotent: false);
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write-split");
await driver.InitializeAsync(string.Empty, ct);
(await driver.WriteAsync([new WriteRequest(rawPath, 7.25)], ct))[0].StatusCode.ShouldBe(Good);
// Positive control first: once the write HAS been seen, the read topic has had its chance.
await writeWatcher.WaitForOneAsync(MessageTimeout, ct);
readWatcher.Count.ShouldBe(
0,
"a write echoed onto the tag's own read topic would read back as a confirmed device value");
}
[Fact]
public async Task Plain_write_to_a_tag_with_no_write_topic_is_refused_end_to_end()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
const string rawPath = "Fixture/Mqtt/Line1/ReadOnly";
var tag = new RawTagEntry(
rawPath,
$$"""{"topic":"{{_fx.Topic("line1/temperature")}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
WriteIdempotent: false);
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write-readonly");
await driver.InitializeAsync(string.Empty, ct);
// Connected and resolvable — so BadNotWritable here is specifically "you never said where",
// not a connection failure wearing the same status.
(await driver.WriteAsync([new WriteRequest(rawPath, 1.0)], ct))[0].StatusCode.ShouldBe(BadNotWritable);
}
// -----------------------------------------------------------------------
// Sparkplug B
// -----------------------------------------------------------------------
[Fact]
public async Task Sparkplug_write_to_a_device_metric_publishes_a_DCMD()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
var group = $"WriteGate{Guid.NewGuid():N}"[..20];
const string node = "EdgeW";
const string device = "Filler1";
const string metric = "Setpoint";
const string rawPath = "Fixture/Mqtt/Sp/Setpoint";
var topic = $"spBv1.0/{group}/DCMD/{node}/{device}";
await using var watcher = await TopicWatcher.StartAsync(_fx, topic, ct);
var tag = new RawTagEntry(
rawPath,
$$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","deviceId":"{{device}}","metricName":"{{metric}}","dataType":"Float64"}""",
WriteIdempotent: false);
await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-dcmd");
await driver.InitializeAsync(string.Empty, ct);
(await driver.WriteAsync([new WriteRequest(rawPath, 12.5)], ct))[0].StatusCode.ShouldBe(Good);
var bytes = await watcher.WaitForOneAsync(MessageTimeout, ct);
var payload = Payload.Parser.ParseFrom(bytes);
payload.Metrics.ShouldHaveSingleItem();
payload.Metrics[0].Name.ShouldBe(metric, "a command must address the metric by NAME, never by alias");
payload.Metrics[0].Datatype.ShouldBe((uint)TahuDataType.Double);
payload.Metrics[0].DoubleValue.ShouldBe(12.5);
payload.HasSeq.ShouldBeFalse("seq belongs to the edge node's own outbound stream, not to a command");
}
[Fact]
public async Task Sparkplug_write_to_an_edge_node_metric_publishes_an_NCMD()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
var group = $"WriteGate{Guid.NewGuid():N}"[..20];
const string node = "EdgeW";
const string metric = "NodeSetpoint";
const string rawPath = "Fixture/Mqtt/Sp/NodeSetpoint";
await using var watcher = await TopicWatcher.StartAsync(_fx, $"spBv1.0/{group}/NCMD/{node}", ct);
var tag = new RawTagEntry(
rawPath,
$$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","metricName":"{{metric}}","dataType":"Int32"}""",
WriteIdempotent: false);
await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-ncmd");
await driver.InitializeAsync(string.Empty, ct);
(await driver.WriteAsync([new WriteRequest(rawPath, 7)], ct))[0].StatusCode.ShouldBe(Good);
// NCMD is a SHARED channel: the driver publishes its own rebirth request
// ("Node Control/Rebirth") on this exact topic when it connects, and the live gate caught it
// arriving first. Wait for the command carrying OUR metric rather than for the first NCMD —
// the DCMD case has no such contention, which is why only this one needed it.
var payload = Payload.Parser.ParseFrom(
await watcher.WaitForMatchAsync(
bytes => Payload.Parser.ParseFrom(bytes).Metrics.Any(m => m.Name == metric),
MessageTimeout,
ct));
var written = payload.Metrics.ShouldHaveSingleItem();
written.Name.ShouldBe(metric);
written.Datatype.ShouldBe((uint)TahuDataType.Int32);
written.IntValue.ShouldBe(7u);
}
[Fact]
public async Task A_sparkplug_command_is_not_retained()
{
SkipIfUnconfigured();
var ct = TestContext.Current.CancellationToken;
var group = $"WriteGate{Guid.NewGuid():N}"[..20];
const string node = "EdgeW";
const string rawPath = "Fixture/Mqtt/Sp/Retain";
var topic = $"spBv1.0/{group}/NCMD/{node}";
await using var firstWatcher = await TopicWatcher.StartAsync(_fx, topic, ct);
var tag = new RawTagEntry(
rawPath,
$$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","metricName":"Cmd","dataType":"Boolean"}""",
WriteIdempotent: false);
await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-retain");
await driver.InitializeAsync(string.Empty, ct);
(await driver.WriteAsync([new WriteRequest(rawPath, true)], ct))[0].StatusCode.ShouldBe(Good);
await firstWatcher.WaitForOneAsync(MessageTimeout, ct);
// A retained command would be replayed by the broker to every LATER subscriber — including the
// device's own reconnect — so a single stale write would be re-applied forever. Subscribing
// fresh, after the publish, is the only way to observe that.
await using var lateWatcher = await TopicWatcher.StartAsync(_fx, topic, ct);
await Task.Delay(TimeSpan.FromSeconds(2), ct);
// Counting OUR command only. The driver's own rebirth NCMD shares this topic, so a bare count
// would make this assertion depend on rebirth timing rather than on the retain flag.
lateWatcher.CountMatching(bytes => Payload.Parser.ParseFrom(bytes).Metrics.Any(m => m.Name == "Cmd"))
.ShouldBe(0, "a retained NCMD/DCMD would drive a device into a permanent command loop");
}
// -----------------------------------------------------------------------
// Helpers
// -----------------------------------------------------------------------
private MqttDriverOptions PlainOptions(params RawTagEntry[] rawTags) => new()
{
Host = _fx.PlainHost,
Port = _fx.PlainPort,
UseTls = false,
Username = _fx.Username,
Password = _fx.Password,
ClientId = $"otopcua-live-{Guid.NewGuid():N}",
ConnectTimeoutSeconds = 15,
Mode = MqttMode.Plain,
Plain = new MqttPlainOptions { DefaultQos = 1 },
RawTags = rawTags,
};
private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new()
{
Host = _fx.PlainHost,
Port = _fx.PlainPort,
UseTls = false,
Username = _fx.Username,
Password = _fx.Password,
ClientId = $"otopcua-live-{Guid.NewGuid():N}",
ConnectTimeoutSeconds = 15,
Mode = MqttMode.SparkplugB,
Sparkplug = new MqttSparkplugOptions { GroupId = groupId },
RawTags = rawTags,
};
/// <summary>
/// An independent broker subscriber that records the payloads seen on one concrete topic — the
/// third party that makes "the write actually reached the broker" an observation rather than an
/// assumption about the driver's own code.
/// </summary>
private sealed class TopicWatcher : IAsyncDisposable
{
private readonly IMqttClient _client;
private readonly ConcurrentQueue<byte[]> _payloads = new();
private TopicWatcher(IMqttClient client) => _client = client;
/// <summary>How many messages have been observed on the watched topic.</summary>
public int Count => _payloads.Count;
public static async Task<TopicWatcher> StartAsync(MqttFixture fixture, string topic, CancellationToken ct)
{
var client = new MqttClientFactory().CreateMqttClient();
var watcher = new TopicWatcher(client);
client.ApplicationMessageReceivedAsync += args =>
{
watcher._payloads.Enqueue(args.ApplicationMessage.Payload.ToArray());
return Task.CompletedTask;
};
var options = new MqttClientOptionsBuilder()
.WithTcpServer(fixture.PlainHost, fixture.PlainPort)
.WithCredentials(fixture.Username, fixture.Password)
.WithClientId($"otopcua-live-watch-{Guid.NewGuid():N}")
.WithTimeout(TimeSpan.FromSeconds(10))
.Build();
await client.ConnectAsync(options, ct).ConfigureAwait(false);
await client.SubscribeAsync(
new MqttClientSubscribeOptionsBuilder().WithTopicFilter(f => f.WithTopic(topic)).Build(),
ct).ConfigureAwait(false);
return watcher;
}
/// <summary>How many observed messages satisfy <paramref name="predicate"/>.</summary>
/// <param name="predicate">The payload test.</param>
/// <returns>The matching count.</returns>
public int CountMatching(Func<byte[], bool> predicate) => _payloads.Count(p => Safe(predicate, p));
/// <summary>
/// Waits for a message satisfying <paramref name="predicate"/>. Needed where the topic is
/// shared — an NCMD topic also carries the driver's own rebirth requests.
/// </summary>
/// <param name="predicate">The payload test.</param>
/// <param name="timeout">Bound on the wait.</param>
/// <param name="ct">Caller cancellation.</param>
/// <returns>The first matching payload.</returns>
public async Task<byte[]> WaitForMatchAsync(Func<byte[], bool> predicate, TimeSpan timeout, CancellationToken ct)
{
var deadline = DateTime.UtcNow + timeout;
while (true)
{
if (_payloads.FirstOrDefault(p => Safe(predicate, p)) is { } match)
{
return match;
}
if (DateTime.UtcNow >= deadline)
{
throw new TimeoutException(
$"No matching message arrived on the watched topic within {timeout} " +
$"({_payloads.Count} message(s) seen, none matching).");
}
await Task.Delay(100, ct).ConfigureAwait(false);
}
}
/// <summary>Applies a payload predicate, treating an undecodable payload as "no match".</summary>
/// <param name="predicate">The payload test.</param>
/// <param name="payload">The observed bytes.</param>
/// <returns><see langword="true"/> when the payload matches.</returns>
private static bool Safe(Func<byte[], bool> predicate, byte[] payload)
{
try { return predicate(payload); }
catch { return false; }
}
public async Task<byte[]> WaitForOneAsync(TimeSpan timeout, CancellationToken ct)
{
var deadline = DateTime.UtcNow + timeout;
while (!_payloads.TryPeek(out _))
{
if (DateTime.UtcNow >= deadline)
{
throw new TimeoutException(
$"No message arrived on the watched topic within {timeout}. The write reported success " +
"but nothing reached the broker — the exact failure this gate exists to catch.");
}
await Task.Delay(100, ct).ConfigureAwait(false);
}
_payloads.TryDequeue(out var payload);
return payload!;
}
public async ValueTask DisposeAsync()
{
try { await _client.DisconnectAsync(cancellationToken: default).ConfigureAwait(false); }
catch { /* teardown must not mask a test failure */ }
_client.Dispose();
}
}
}
@@ -410,7 +410,7 @@ public sealed class MqttBrowseSessionTests
a.Length.ShouldBe("-browse-".Length + 8);
var opts = new MqttDriverOptions { ClientId = "otopcua-node1" };
MqttConnection.BuildClientOptions(opts, a).ClientId.ShouldBe("otopcua-node1" + a);
MqttClientOptionsFactory.BuildClientOptions(opts, a).ClientId.ShouldBe("otopcua-node1" + a);
}
[Fact]
@@ -0,0 +1,85 @@
using System.Reflection;
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
/// <summary>
/// Pins the MQTT project layering that Gitea #512 restored: <c>.Browser</c> depends on the shared
/// <c>.Contracts</c> + <c>.Transport</c> leaves, never on the runtime <c>.Driver</c>.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why this is worth a test.</b> <c>.Browser</c> used to reference the runtime driver, and the
/// AdminUI references <c>.Browser</c> — so the AdminUI's build/deploy graph transitively contained
/// the whole MQTT driver runtime plus Core, Polly.Core and Serilog. Nothing failed; the cost was
/// invisible. A reference re-added out of convenience would be invisible in exactly the same way,
/// which is precisely what a guard is for.
/// </para>
/// <para>
/// <b>What it can and cannot catch.</b> It asserts on emitted assembly references, so it fires when
/// the browser actually <em>uses</em> a driver-runtime type — the case that matters. A
/// <c>ProjectReference</c> re-added but never used would not trip it (the compiler emits no
/// reference), so the csproj comment remains the primary record of the decision.
/// </para>
/// </remarks>
[Trait("Category", "Unit")]
public sealed class MqttBrowserLayeringTests
{
private const string BrowserAssembly = "ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser";
private const string DriverRuntimeAssembly = "ZB.MOM.WW.OtOpcUa.Driver.Mqtt";
private const string ContractsAssembly = "ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts";
private const string TransportAssembly = "ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport";
private static string[] BrowserReferences()
{
var browser = Assembly.Load(BrowserAssembly);
return [.. browser.GetReferencedAssemblies().Select(a => a.Name!)];
}
[Fact]
public void Browser_does_not_reference_the_runtime_driver()
=> BrowserReferences().ShouldNotContain(
DriverRuntimeAssembly,
"the MQTT browser is back to referencing the runtime driver — every other bespoke browser " +
"in this repo (OpcUaClient, Galaxy) depends on its .Contracts leaf only, and this reference " +
"reaches the AdminUI. If shared connection policy is needed, put it in .Transport; if shared " +
"protocol/codec is needed, put it in .Contracts. Do NOT copy the TLS/CA-pin logic.");
[Fact]
public void Browser_reaches_the_shared_leaves_instead()
{
// The positive control. Asserting only the absence above would also pass if the browser stopped
// referencing anything MQTT at all — e.g. if this test loaded the wrong assembly, or the type
// names drifted — so pin what it SHOULD depend on in the same breath.
var references = BrowserReferences();
references.ShouldContain(ContractsAssembly);
references.ShouldContain(TransportAssembly);
}
[Fact]
public void Transport_stays_a_leaf_that_does_not_reach_back_into_the_runtime_driver()
{
// .Transport holds connection POLICY only. If it ever referenced the runtime driver, routing the
// browser through it would restore the coupling this whole change removed, one hop further out.
var references = Assembly.Load(TransportAssembly)
.GetReferencedAssemblies()
.Select(a => a.Name!)
.ToArray();
references.ShouldNotContain(DriverRuntimeAssembly);
references.ShouldNotContain(BrowserAssembly);
}
[Fact]
public void Contracts_stays_transport_free()
=> Assembly.Load(ContractsAssembly)
.GetReferencedAssemblies()
.Select(a => a.Name!)
.ShouldNotContain(
"MQTTnet",
"Driver.Mqtt.Contracts is deliberately transport-free — it carries the Sparkplug wire " +
"schema and codec, and is referenced by ControlPlane, which has no business with a " +
"broker client. MQTTnet-typed connection policy belongs in .Transport.");
}
@@ -13,7 +13,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
/// <summary>
/// Covers <see cref="MqttConnection"/>'s two Task-3 responsibilities: the pure
/// <see cref="MqttConnection.BuildClientOptions"/> options assembly (TLS / CA pin /
/// <see cref="MqttClientOptionsFactory.BuildClientOptions"/> options assembly (TLS / CA pin /
/// credentials / protocol knobs) and the bounded <see cref="MqttConnection.ConnectAsync"/>
/// deadline.
/// </summary>
@@ -189,7 +189,7 @@ public sealed class MqttConnectionTests
var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true, CaCertificatePath = missing };
var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
var built = MqttClientOptionsFactory.BuildClientOptions(opts, clientIdSuffix: null);
var tls = built.ChannelOptions.ShouldBeOfType<MqttClientTcpOptions>().TlsOptions;
tls.CertificateValidationHandler.ShouldNotBeNull();
@@ -206,7 +206,7 @@ public sealed class MqttConnectionTests
try
{
var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true, CaCertificatePath = caPath };
var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
var built = MqttClientOptionsFactory.BuildClientOptions(opts, clientIdSuffix: null);
var tls = built.ChannelOptions.ShouldBeOfType<MqttClientTcpOptions>().TlsOptions;
using var stranger = SelfSignedLeaf(Host);
@@ -232,7 +232,7 @@ public sealed class MqttConnectionTests
try
{
var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true, CaCertificatePath = caPath };
var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
var built = MqttClientOptionsFactory.BuildClientOptions(opts, clientIdSuffix: null);
var tls = built.ChannelOptions.ShouldBeOfType<MqttClientTcpOptions>().TlsOptions;
tls.CertificateValidationHandler!(ValidationArgs(built, leaf)).ShouldBeTrue();
@@ -259,7 +259,7 @@ public sealed class MqttConnectionTests
try
{
var opts = new MqttDriverOptions { Host = Host, Port = 8883, UseTls = true, CaCertificatePath = caPath };
var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
var built = MqttClientOptionsFactory.BuildClientOptions(opts, clientIdSuffix: null);
var tls = built.ChannelOptions.ShouldBeOfType<MqttClientTcpOptions>().TlsOptions;
tls.CertificateValidationHandler!(ValidationArgs(built, leaf, intermediate)).ShouldBeTrue();
@@ -330,7 +330,7 @@ public sealed class MqttConnectionTests
ConnectTimeoutSeconds = 7,
};
var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
var built = MqttClientOptionsFactory.BuildClientOptions(opts, clientIdSuffix: null);
built.ProtocolVersion.ShouldBe(MQTTnet.Formatter.MqttProtocolVersion.V311);
built.ClientId.ShouldBe("otopcua-node-1");
@@ -347,12 +347,12 @@ public sealed class MqttConnectionTests
{
var opts = new MqttDriverOptions { Host = Host, ClientId = "otopcua" };
MqttConnection.BuildClientOptions(opts, clientIdSuffix: "-browse").ClientId.ShouldBe("otopcua-browse");
MqttClientOptionsFactory.BuildClientOptions(opts, clientIdSuffix: "-browse").ClientId.ShouldBe("otopcua-browse");
}
[Fact]
public void BuildClientOptions_NoUsername_SendsNoCredentials()
=> MqttConnection.BuildClientOptions(new MqttDriverOptions { Host = Host }, clientIdSuffix: null)
=> MqttClientOptionsFactory.BuildClientOptions(new MqttDriverOptions { Host = Host }, clientIdSuffix: null)
.Credentials.ShouldBeNull();
[Fact]
@@ -360,7 +360,7 @@ public sealed class MqttConnectionTests
{
var opts = new MqttDriverOptions { Host = Host, Username = "svc", Password = "pw" };
var built = MqttConnection.BuildClientOptions(opts, clientIdSuffix: null);
var built = MqttClientOptionsFactory.BuildClientOptions(opts, clientIdSuffix: null);
built.Credentials.ShouldNotBeNull();
built.Credentials.GetUserName(built).ShouldBe("svc");
@@ -1166,7 +1166,7 @@ public sealed class MqttConnectionTests
}
private static MqttClientTlsOptions BuildTls(MqttDriverOptions opts)
=> MqttConnection.BuildClientOptions(opts, clientIdSuffix: null)
=> MqttClientOptionsFactory.BuildClientOptions(opts, clientIdSuffix: null)
.ChannelOptions.ShouldBeOfType<MqttClientTcpOptions>().TlsOptions;
private static readonly DateTimeOffset NotBefore = DateTimeOffset.UtcNow.AddDays(-1);
@@ -0,0 +1,160 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
/// <summary>
/// Unit tests for <see cref="MqttDriver"/>'s <see cref="IWritable"/> leg (Gitea #508) — the batch
/// contract and the per-slot refusals, up to (not including) the broker publish.
/// </summary>
/// <remarks>
/// The publish itself is the live gate's job: it needs a broker, and asserting it here would only
/// prove that a fake was called. What IS worth pinning offline is everything that decides whether a
/// write should reach a broker at all — because that is where a silent no-op would hide.
/// </remarks>
[Trait("Category", "Unit")]
public sealed class MqttDriverWriteTests
{
private const string Path = "Plant/Mqtt/dev1/Setpoint";
private const string ReadOnlyPath = "Plant/Mqtt/dev1/Temp";
private static RawTagEntry PlainTag(string rawPath, string? writeTopic)
{
var writeKey = writeTopic is null ? "" : $$""","writeTopic":"{{writeTopic}}" """;
return new RawTagEntry(
rawPath,
$$"""{"topic":"factory/oven/temp","payloadFormat":"Scalar","dataType":"Float64"{{writeKey}}}""",
WriteIdempotent: false);
}
private static MqttDriver PlainDriver(params RawTagEntry[] tags)
=> new(new MqttDriverOptions { Mode = MqttMode.Plain, RawTags = tags }, "d", null);
[Fact]
public void MqttDriver_implements_IWritable()
// The interface's presence is the feature: without it every MQTT node materialises read-only,
// regardless of what the operator authored on the tag.
=> PlainDriver().ShouldBeAssignableTo<IWritable>();
[Fact]
public async Task An_unknown_reference_is_refused_in_its_own_slot()
{
var results = await PlainDriver(PlainTag(Path, "oven/set"))
.WriteAsync([new WriteRequest("Plant/Mqtt/dev1/NotATag", 1.0)], default);
results.Count.ShouldBe(1);
results[0].StatusCode.ShouldBe(0x80340000u); // BadNodeIdUnknown
}
[Fact]
public async Task A_tag_with_no_write_topic_is_refused_as_not_writable()
{
// NOT BadCommunicationError and NOT Good: the tag is addressable and the broker is irrelevant —
// the operator simply never said where a write should go. The status has to say that, or the
// only diagnosis available is "writes don't work".
var results = await PlainDriver(PlainTag(ReadOnlyPath, writeTopic: null))
.WriteAsync([new WriteRequest(ReadOnlyPath, 1.0)], default);
results[0].StatusCode.ShouldBe(0x803B0000u); // BadNotWritable
}
[Fact]
public async Task A_writable_tag_on_a_disconnected_driver_reports_a_communication_failure()
{
// Reported, never queued. A write buffered now and flushed at an unbounded later moment is the
// replay hazard WriteIdempotent exists to prevent — and this would be the driver deciding it on
// the operator's behalf.
var results = await PlainDriver(PlainTag(Path, "oven/set"))
.WriteAsync([new WriteRequest(Path, 42.5)], default);
results[0].StatusCode.ShouldBe(0x80050000u); // BadCommunicationError
}
[Fact]
public async Task One_bad_request_does_not_fail_the_others_in_the_batch()
{
// Mirrors ReadAsync: failing the whole call would let one stale reference reject every other
// write in the same OPC UA request.
var driver = PlainDriver(PlainTag(Path, "oven/set"), PlainTag(ReadOnlyPath, writeTopic: null));
var results = await driver.WriteAsync(
[
new WriteRequest("Plant/Mqtt/dev1/NotATag", 1.0),
new WriteRequest(ReadOnlyPath, 1.0),
new WriteRequest(Path, 42.5),
],
default);
// Order-preserving, one status per request, each naming its own distinct cause.
results.Count.ShouldBe(3);
results[0].StatusCode.ShouldBe(0x80340000u); // BadNodeIdUnknown
results[1].StatusCode.ShouldBe(0x803B0000u); // BadNotWritable
results[2].StatusCode.ShouldBe(0x80050000u); // BadCommunicationError — resolved and encodable
}
[Fact]
public async Task An_empty_batch_is_answered_with_an_empty_result()
=> (await PlainDriver().WriteAsync([], default)).ShouldBeEmpty();
[Fact]
public async Task WriteAsync_rejects_a_null_batch()
=> await Should.ThrowAsync<ArgumentNullException>(
async () => await PlainDriver().WriteAsync(null!, default));
// ---- Discovery advertises writability consistently with the write path ------------------------
[Fact]
public async Task A_tag_with_a_write_topic_is_discovered_as_an_operate_node()
{
var builder = new WriteCapturingBuilder();
await PlainDriver(PlainTag(Path, "oven/set")).DiscoverAsync(builder, default);
builder.Variables.ShouldHaveSingleItem();
builder.Variables[0].SecurityClass.ShouldBe(SecurityClassification.Operate);
// Default-off regardless: MQTT gives no application-level write ack, and a Sparkplug command is
// not necessarily idempotent. Opting in is the operator's call, per tag.
builder.Variables[0].WriteIdempotent.ShouldBeFalse();
}
[Fact]
public async Task A_tag_without_a_write_topic_stays_view_only()
{
// The browse-time answer must match the runtime one — a node advertised Operate that then
// refuses every write would be the silent-no-op failure in a new costume.
var builder = new WriteCapturingBuilder();
await PlainDriver(PlainTag(ReadOnlyPath, writeTopic: null)).DiscoverAsync(builder, default);
builder.Variables[0].SecurityClass.ShouldBe(SecurityClassification.ViewOnly);
}
/// <summary>Minimal capturing builder — this project does not reference Commons' own.</summary>
private sealed class WriteCapturingBuilder : IAddressSpaceBuilder
{
public List<DriverAttributeInfo> Variables { get; } = [];
public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;
public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
{
Variables.Add(attributeInfo);
return new Handle(attributeInfo.FullName);
}
public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
private sealed class Handle(string fullRef) : IVariableHandle
{
public string FullReference => fullRef;
public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => new NullSink();
}
private sealed class NullSink : IAlarmConditionSink
{
public void OnTransition(AlarmEventArgs args) { }
}
}
}
@@ -0,0 +1,251 @@
using System.Text;
using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
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.Tests;
/// <summary>
/// Unit tests for <see cref="MqttWriteEncoder"/> — the pure half of MQTT write-through (Gitea #508):
/// what a write encodes to, and which writes are refused before anything reaches a broker.
/// </summary>
[Trait("Category", "Unit")]
public sealed class MqttWriteEncoderTests
{
private static MqttTagDefinition Plain(
string? writeTopic,
MqttPayloadFormat format = MqttPayloadFormat.Scalar,
DriverDataType dataType = DriverDataType.Float64)
=> new(
Name: "Plant/Mqtt/dev1/Setpoint",
Topic: "factory/oven/temp",
PayloadFormat: format,
JsonPath: "$",
DataType: dataType,
Qos: null,
RetainSeed: true,
WriteTopic: writeTopic);
private static MqttTagDefinition Sparkplug(
string? deviceId,
DriverDataType dataType = DriverDataType.Float64,
string metricName = "Speed")
=> new(
Name: "Plant/Mqtt/dev1/Speed",
Topic: "",
PayloadFormat: MqttPayloadFormat.Json,
JsonPath: "$",
DataType: dataType,
Qos: null,
RetainSeed: true,
GroupId: "OtOpcUaSim",
EdgeNodeId: "EdgeA",
DeviceId: deviceId,
MetricName: metricName);
// ---- The refusal that is the whole point of the feature -------------------------------------
[Fact]
public void A_plain_tag_with_no_write_topic_is_refused_rather_than_published_nowhere()
{
// Before #508 there was no IWritable at all — honest read-only. The one outcome that would be
// WORSE is a node that accepts a write, reports Good and publishes nothing. So this must be a
// refusal, and specifically the "no destination" one, which the driver maps to BadNotWritable.
var error = MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain(writeTopic: null), 42.5, out var encoded);
error.ShouldBe(MqttWriteEncodeError.NoWriteDestination);
encoded.Topic.ShouldBeNull();
encoded.Payload.ShouldBeNull();
}
[Theory]
[InlineData("")]
[InlineData(" ")]
public void A_blank_write_topic_is_treated_as_absent(string writeTopic)
=> MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain(writeTopic), 1, out _)
.ShouldBe(MqttWriteEncodeError.NoWriteDestination);
[Fact]
public void A_sparkplug_tag_with_an_incomplete_tuple_has_no_command_address()
{
// A Sparkplug tag's command address IS its binding tuple, so an incomplete one is the same
// failure as a missing write topic, not a different kind.
var def = Sparkplug(deviceId: null) with { MetricName = null };
MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 1.0, out _)
.ShouldBe(MqttWriteEncodeError.NoWriteDestination);
}
// ---- Plain encoding --------------------------------------------------------------------------
[Fact]
public void Plain_publishes_to_the_write_topic_not_the_read_topic()
{
// The two-field split is the point: publishing a command onto the tag's own read topic would
// echo back and read as a confirmed device value.
var def = Plain("factory/oven/temp/set");
MqttWriteEncoder.TryEncode(MqttMode.Plain, def, 42.5, out var encoded)
.ShouldBe(MqttWriteEncodeError.None);
encoded.Topic.ShouldBe("factory/oven/temp/set");
encoded.Topic.ShouldNotBe(def.Topic);
}
[Theory]
[InlineData(MqttPayloadFormat.Scalar)]
[InlineData(MqttPayloadFormat.Raw)]
public void Scalar_and_raw_encode_the_value_as_utf8_text(MqttPayloadFormat format)
{
MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("t/set", format), 42.5, out var encoded)
.ShouldBe(MqttWriteEncodeError.None);
Encoding.UTF8.GetString(encoded.Payload).ShouldBe("42.5");
}
[Fact]
public void Numbers_are_encoded_with_the_invariant_culture()
{
// A comma decimal separator would publish a number the device cannot parse — and it would only
// reproduce on a host with a European locale, which is the worst kind of bug to chase.
var previous = Thread.CurrentThread.CurrentCulture;
Thread.CurrentThread.CurrentCulture = new System.Globalization.CultureInfo("de-DE");
try
{
MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("t/set"), 42.5, out var encoded)
.ShouldBe(MqttWriteEncodeError.None);
Encoding.UTF8.GetString(encoded.Payload).ShouldBe("42.5");
}
finally
{
Thread.CurrentThread.CurrentCulture = previous;
}
}
[Fact]
public void Json_encodes_a_bare_json_literal()
{
MqttWriteEncoder.TryEncode(
MqttMode.Plain, Plain("t/set", MqttPayloadFormat.Json), 42.5, out var encoded)
.ShouldBe(MqttWriteEncodeError.None);
Encoding.UTF8.GetString(encoded.Payload).ShouldBe("42.5");
}
[Fact]
public void Json_quotes_a_string_value()
=> Encoding.UTF8.GetString(
EncodeOrThrow(MqttMode.Plain, Plain("t/set", MqttPayloadFormat.Json, DriverDataType.String), "on"))
.ShouldBe("\"on\"");
[Fact]
public void A_null_value_is_refused()
// A zero-length MQTT payload is a legitimate clear-the-retained-message instruction — an
// entirely different operation. Encoding null as "no bytes" would conflate the two.
=> MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("t/set"), null, out _)
.ShouldBe(MqttWriteEncodeError.ValueNotRepresentable);
// ---- Sparkplug command encoding ---------------------------------------------------------------
[Fact]
public void A_device_metric_encodes_as_a_DCMD_addressed_to_that_device()
{
// Addressing a device metric with NCMD would command the NODE, which does not own it.
MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, Sparkplug("Filler1"), 12.5, out var encoded)
.ShouldBe(MqttWriteEncodeError.None);
encoded.Topic.ShouldBe("spBv1.0/OtOpcUaSim/DCMD/EdgeA/Filler1");
}
[Fact]
public void An_edge_node_metric_encodes_as_an_NCMD()
=> EncodeTopic(MqttMode.SparkplugB, Sparkplug(deviceId: null))
.ShouldBe("spBv1.0/OtOpcUaSim/NCMD/EdgeA");
[Fact]
public void The_command_carries_the_metric_by_name_with_an_explicit_datatype()
{
// By NAME, never by alias: an alias may be reused for a different metric across a rebirth.
// Datatype explicitly, because a node that treats it as ambiguous drops the command SILENTLY —
// the exact failure mode this feature must not have.
var bytes = EncodeOrThrow(MqttMode.SparkplugB, Sparkplug("Filler1"), 12.5);
var payload = Payload.Parser.ParseFrom(bytes);
payload.Metrics.Count.ShouldBe(1);
var metric = payload.Metrics[0];
metric.Name.ShouldBe("Speed");
metric.Alias.ShouldBe(0ul);
metric.Datatype.ShouldBe((uint)TahuDataType.Double);
metric.DoubleValue.ShouldBe(12.5);
metric.Timestamp.ShouldBeGreaterThan(0ul);
payload.Timestamp.ShouldBeGreaterThan(0ul);
}
[Fact]
public void The_command_carries_no_sequence_number()
{
// seq belongs to the messages an edge node itself publishes. A strict node validating a
// host-supplied one against its own counter would have grounds to reject the command.
var payload = Payload.Parser.ParseFrom(EncodeOrThrow(MqttMode.SparkplugB, Sparkplug("Filler1"), 1.0));
payload.HasSeq.ShouldBeFalse();
}
[Theory]
[InlineData(DriverDataType.Boolean, true, (uint)TahuDataType.Boolean)]
[InlineData(DriverDataType.Int32, 7, (uint)TahuDataType.Int32)]
[InlineData(DriverDataType.Int64, 7L, (uint)TahuDataType.Int64)]
[InlineData(DriverDataType.UInt32, 7, (uint)TahuDataType.Uint32)]
[InlineData(DriverDataType.Float32, 1.5f, (uint)TahuDataType.Float)]
[InlineData(DriverDataType.String, "x", (uint)TahuDataType.String)]
public void The_declared_data_type_selects_the_wire_datatype(
DriverDataType dataType, object value, uint expectedWireType)
{
var payload = Payload.Parser.ParseFrom(
EncodeOrThrow(MqttMode.SparkplugB, Sparkplug("Filler1", dataType), value));
payload.Metrics[0].Datatype.ShouldBe(expectedWireType);
}
[Fact]
public void An_unrepresentable_value_is_refused_rather_than_coerced()
// "not-a-number" into a Float64 tag must not silently become 0 on a live plant.
=> MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, Sparkplug("Filler1"), "not-a-number", out _)
.ShouldBe(MqttWriteEncodeError.ValueNotRepresentable);
[Fact]
public void A_type_with_no_command_encoding_is_refused_rather_than_stringified()
// Reference has no Sparkplug wire representation. Falling back to String would put a value on
// the wire that is not what the tag's type says it is.
=> MqttWriteEncoder.TryEncode(
MqttMode.SparkplugB, Sparkplug("Filler1", DriverDataType.Reference), "ref", out _)
.ShouldBe(MqttWriteEncodeError.UnsupportedDataType);
[Fact]
public void The_mode_decides_the_shape_not_the_blob()
{
// A tag can legitimately carry both shapes' keys (the editor preserves unknown keys across a
// retype), so sniffing would make one authored tag mean two different things.
var both = Sparkplug("Filler1") with { WriteTopic = "plain/set" };
EncodeTopic(MqttMode.Plain, both).ShouldBe("plain/set");
EncodeTopic(MqttMode.SparkplugB, both).ShouldBe("spBv1.0/OtOpcUaSim/DCMD/EdgeA/Filler1");
}
private static string EncodeTopic(MqttMode mode, MqttTagDefinition def)
{
MqttWriteEncoder.TryEncode(mode, def, 1.0, out var encoded).ShouldBe(MqttWriteEncodeError.None);
return encoded.Topic;
}
private static byte[] EncodeOrThrow(MqttMode mode, MqttTagDefinition def, object value)
{
MqttWriteEncoder.TryEncode(mode, def, value, out var encoded).ShouldBe(MqttWriteEncodeError.None);
return encoded.Payload;
}
}
@@ -0,0 +1,84 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.AdminUI.Browsing;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Browsing;
/// <summary>
/// Unit tests for <see cref="SparkplugRebirthScope"/> — the operator-typed <c>{group}/{edgeNode}</c>
/// rebirth scope that breaks the birth-driven-tree deadlock (Gitea #514).
/// </summary>
/// <remarks>
/// The rule is deliberately advisory — <c>MqttBrowseSession.ResolveRebirthTargets</c> re-resolves the
/// scope and refuses an unusable one before publishing — so these tests pin the <em>affordance</em>
/// behaviour: an operator must not be able to arm a two-click confirm that cannot succeed, and must
/// not be blocked from a scope the session would accept.
/// </remarks>
[Trait("Category", "Unit")]
public sealed class SparkplugRebirthScopeTests
{
[Theory]
[InlineData("OtOpcUaSim/EdgeA")]
[InlineData(" OtOpcUaSim/EdgeA ")]
[InlineData("g/e")]
[InlineData("Plant 1/Edge Node 2")] // spaces are legal inside a Sparkplug id
public void Accepts_a_group_and_edge_node(string scope)
{
SparkplugRebirthScope.Describe(scope).ShouldBeNull();
SparkplugRebirthScope.IsUsable(scope).ShouldBeTrue();
}
[Theory]
[InlineData(null)]
[InlineData("")]
[InlineData(" ")]
public void Rejects_a_blank_scope(string? scope)
=> SparkplugRebirthScope.Describe(scope).ShouldNotBeNullOrWhiteSpace();
[Fact]
public void Rejects_a_group_on_its_own()
{
// The session resolves a bare group by enumerating OBSERVED edge nodes, of which an empty tree
// has none — so accepting it here would arm a confirm step that resolves to nothing, in exactly
// the situation this input exists for.
var error = SparkplugRebirthScope.Describe("OtOpcUaSim");
error.ShouldNotBeNullOrWhiteSpace();
error!.ShouldContain("group/edgeNode");
}
[Theory]
[InlineData("g/e/extra")]
[InlineData("g//e")]
[InlineData("/EdgeA")]
[InlineData("OtOpcUaSim/")]
public void Rejects_a_scope_that_is_not_exactly_two_segments(string scope)
=> SparkplugRebirthScope.Describe(scope).ShouldNotBeNullOrWhiteSpace();
[Theory]
[InlineData("g/+")]
[InlineData("#/e")]
[InlineData("g/Edge#")]
public void Rejects_an_mqtt_wildcard(string scope)
{
// A rebirth NCMD is addressed to one edge node on a literal topic; a wildcard names no single
// node, and letting one through would build a topic the broker rejects.
var error = SparkplugRebirthScope.Describe(scope);
error.ShouldNotBeNullOrWhiteSpace();
error!.ShouldContain("wildcard");
}
[Fact]
public void Split_returns_the_trimmed_pair()
{
var (groupId, edgeNodeId) = SparkplugRebirthScope.Split(" OtOpcUaSim/EdgeA ");
groupId.ShouldBe("OtOpcUaSim");
edgeNodeId.ShouldBe("EdgeA");
}
[Fact]
public void Split_refuses_a_scope_Describe_would_reject()
=> Should.Throw<ArgumentException>(() => SparkplugRebirthScope.Split("OtOpcUaSim"));
}
@@ -160,6 +160,58 @@ public sealed class MqttTagConfigModelTests
json.ShouldNotContain("FullName", Case.Sensitive);
}
// ── payloadFormat is Plain-only (Gitea #513) ────────────────────────────────────────────────
//
// payloadFormat (Json/Raw/Scalar) selects a PLAIN-mode decode path. MqttTagDefinitionFactory routes
// a Sparkplug tag on its descriptor keys and never reads it, so a payloadFormat in a Sparkplug blob
// is not mis-binding anything — it is a key that looks like it selects a decode path and does not,
// in a blob the operator reads (the raw-JSON textarea, a config export). It was written because
// PayloadFormat is the one non-nullable Plain typed field, so "write what the field says" INVENTED
// the key rather than preserving one.
[Fact]
public void ToJson_omits_payloadFormat_for_a_sparkplug_tag()
{
var json = MqttTagConfigModel.FromJson(
"""{"groupId":"Plant1","edgeNodeId":"EdgeA","metricName":"Node Control/Rebirth"}""").ToJson();
JsonNode.Parse(json)!.AsObject().ContainsKey("payloadFormat").ShouldBeFalse(
"payloadFormat is a Plain-mode decode selector; a Sparkplug tag that never carried one must " +
"not gain it on save. Actual blob: " + json);
}
[Fact]
public void ToJson_still_writes_payloadFormat_for_a_plain_tag()
{
// The negative control for the test above: the omission must be mode-conditional, not a blanket
// removal. A Plain tag's decode path is genuinely selected by this key.
var json = new MqttTagConfigModel { Mode = MqttMode.Plain, Topic = "a/b", PayloadFormat = MqttPayloadFormat.Raw }
.ToJson();
JsonNode.Parse(json)!.AsObject()["payloadFormat"]!.GetValue<string>().ShouldBe("Raw");
}
[Fact]
public void ToJson_clears_a_stale_payloadFormat_when_a_plain_tag_is_retyped_to_sparkplug()
{
// The retype path is how a real blob acquires the stale key. Clearing it is legitimate precisely
// because payloadFormat is a key this editor OWNS — the unknown-key assertions below pin that the
// clearing stays surgical and does not start dropping foreign keys (the issue's explicit warning).
var json = MqttTagConfigModel.FromJson(
"""
{"topic":"a/b","payloadFormat":"Raw","groupId":"Plant1","edgeNodeId":"EdgeA","metricName":"Temp",
"isHistorized":true,"someFutureScalar":"keep me"}
""").ToJson();
var o = JsonNode.Parse(json)!.AsObject();
o.ContainsKey("payloadFormat").ShouldBeFalse();
// Descriptor keys intact, and the keys this editor does not own are untouched.
o["metricName"]!.GetValue<string>().ShouldBe("Temp");
o["isHistorized"]!.GetValue<bool>().ShouldBeTrue();
o["someFutureScalar"]!.GetValue<string>().ShouldBe("keep me");
}
// ── Validate: Sparkplug rules (Task 24) ─────────────────────────────────────────────────────
//
// Mirrors MqttTagDefinitionFactory.FromSparkplugTagConfig exactly: groupId/edgeNodeId/metricName
@@ -3,6 +3,7 @@ using System.Text.Json.Nodes;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.AdminUI.Uns;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
@@ -397,4 +398,128 @@ public sealed class RawBrowseCommitMapperTests
row.Tag.DataType.ShouldBe("Double");
JsonNode.Parse(row.Tag.TagConfig)!.AsObject()["nodeId"]!.GetValue<string>().ShouldBe("ns=2;s=Tag1");
}
// ---- ToRawSegment / name sanitisation (Gitea #509) --------------------------------------------
//
// A tag Name is a RawPath segment and may not contain '/'; a Sparkplug metric name legitimately
// may, and the spec-MANDATORY 'Node Control/Rebirth' does. Committing used to be refused whole.
// The address is a separate field and must survive the coercion untouched.
[Theory]
[InlineData("Node Control/Rebirth", "Node Control_Rebirth")]
[InlineData("Properties/Hardware Make", "Properties_Hardware Make")]
[InlineData("a/b/c", "a_b_c")]
[InlineData(" padded ", "padded")]
[InlineData("plain", "plain")]
public void ToRawSegment_coerces_a_browse_name_into_a_legal_segment(string browseName, string expected)
=> RawBrowseCommitMapper.ToRawSegment(browseName).ShouldBe(expected);
[Theory]
[InlineData("Node Control/Rebirth")]
[InlineData("/leading")]
[InlineData("trailing/")]
[InlineData(" /spaced/ ")]
[InlineData("a/b/c/d/e")]
[InlineData("no-change-needed")]
public void ToRawSegment_output_satisfies_the_RawPaths_segment_rule(string browseName)
{
// Pinned against the AUTHORITY, not against a charset copied into this test. RawPaths.ValidateSegment
// is the rule the import will apply; if it gains a rule the sanitiser does not implement, this fails
// here rather than letting an illegal Name reach ImportTagsAsync.
var segment = RawBrowseCommitMapper.ToRawSegment(browseName);
RawPaths.ValidateSegment(segment).ShouldBeNull(
$"'{browseName}' sanitised to '{segment}', which RawPaths still rejects");
}
[Fact]
public void MapLeaf_commits_a_slashed_sparkplug_metric_with_the_exact_metricName_preserved()
{
var row = RawBrowseCommitMapper.MapLeaf(
driverType: DriverTypeNames.Mqtt,
fullName: "OtOpcUaSim/EdgeA::Node Control/Rebirth",
browseName: "Node Control/Rebirth",
driverDataType: "Boolean",
defaultDataType: "String",
groupPrefix: null,
folderPath: null,
createGroups: false,
addressFields: new Dictionary<string, string>
{
[MqttTagConfigKeys.GroupId] = "OtOpcUaSim",
[MqttTagConfigKeys.EdgeNodeId] = "EdgeA",
[MqttTagConfigKeys.MetricName] = "Node Control/Rebirth",
});
// The Name is coerced...
row.Tag.Name.ShouldBe("Node Control_Rebirth");
RawPaths.ValidateSegment(row.Tag.Name).ShouldBeNull();
// ...and the binding address is NOT. This is the whole point: the two fields already existed
// separately, and only the Name derivation conflated them.
var o = JsonNode.Parse(row.Tag.TagConfig)!.AsObject();
o[MqttTagConfigKeys.MetricName]!.GetValue<string>().ShouldBe("Node Control/Rebirth");
o[MqttTagConfigKeys.GroupId]!.GetValue<string>().ShouldBe("OtOpcUaSim");
o[MqttTagConfigKeys.EdgeNodeId]!.GetValue<string>().ShouldBe("EdgeA");
}
// ---- DescribeNameCollisions ------------------------------------------------------------------
[Fact]
public void DescribeNameCollisions_refuses_two_metrics_that_sanitise_onto_one_name()
{
// 'A/B' and 'A_B' are distinct metrics that both land on 'A_B'. Silently merging them would be
// worse than the refusal #509 replaced — one tag bound to whichever address survived.
var names = new[] { "A/B", "A_B" };
var rows = names
.Select(n => RawBrowseCommitMapper.MapLeaf(
DriverTypeNames.Mqtt, $"g/e::{n}", n, "Float32", "Double", null, null, false,
new Dictionary<string, string>
{
[MqttTagConfigKeys.GroupId] = "g",
[MqttTagConfigKeys.EdgeNodeId] = "e",
[MqttTagConfigKeys.MetricName] = n,
}))
.ToList();
var errors = RawBrowseCommitMapper.DescribeNameCollisions(names, rows);
errors.Count.ShouldBe(1);
// The message must name the specific pair — an anonymous batch failure is what made the original
// refusal hard to act on.
errors[0].ShouldContain("A_B");
errors[0].ShouldContain("'A/B'");
}
[Fact]
public void DescribeNameCollisions_allows_the_same_metric_name_under_two_different_groups()
{
// Uniqueness is per target group, so this is NOT a collision. Without the scoping, mirroring a
// browse tree with a repeated metric name under two devices would be refused wholesale.
var rows = new[]
{
RawBrowseCommitMapper.MapLeaf(
DriverTypeNames.OpcUaClient, "ns=2;s=1", "Temp", null, "Double",
groupPrefix: "DeviceA", folderPath: null, createGroups: false),
RawBrowseCommitMapper.MapLeaf(
DriverTypeNames.OpcUaClient, "ns=2;s=2", "Temp", null, "Double",
groupPrefix: "DeviceB", folderPath: null, createGroups: false),
};
RawBrowseCommitMapper.DescribeNameCollisions(new[] { "Temp", "Temp" }, rows).ShouldBeEmpty();
}
[Fact]
public void DescribeNameCollisions_passes_a_clean_selection()
=> RawBrowseCommitMapper.DescribeNameCollisions(
new[] { "Node Control/Rebirth", "Temperature" },
new[]
{
RawBrowseCommitMapper.MapLeaf(
DriverTypeNames.OpcUaClient, "ns=2;s=1", "Node Control/Rebirth", null, "Double",
null, null, false),
RawBrowseCommitMapper.MapLeaf(
DriverTypeNames.OpcUaClient, "ns=2;s=2", "Temperature", null, "Double",
null, null, false),
}).ShouldBeEmpty();
}