Files
lmxopcua/docs/research/drivers/mqtt-sparkplug.md
T
Joseph Doherty 8fc147d8d4
v2-ci / build (push) Successful in 3m14s
v2-ci / unit-tests (push) Failing after 9m13s
docs: driver-expansion program — 8 research reports + 7 design docs, parallel-reviewed
Adds the driver-expansion program design (umbrella: universal Discover-backed
browser + MTConnect, MQTT/Sparkplug B, BACnet/IP, SQL poll, Omron, Modbus RTU;
MELSEC deferred) plus the per-driver research reports.

All docs went through a 7-agent parallel review against the codebase before
this commit. Highlights fixed in review:

- universal browser: FOCAS FixedTree fills post-connect -> UntilStable settle
  + FixedTree.Enabled patch; MQTT reconciled to bespoke (was contradicting the
  program doc's SupportsOnlineDiscovery=false verdict)
- modbus-rtu: SerialPort.ReadTimeout doesn't bound async BaseStream reads ->
  linked-CTS per-op deadline (R2-01 class); BCL enum reuse would leak
  System.IO.Ports into Contracts
- bacnet: DiscoveryRediscoverPolicy enum name; UDP 47808 contention; live
  suite rewritten around unicast Who-Is + BBMD (broadcast doesn't cross VMs)
- sql-poll: real tier registration via DriverFactoryRegistry.Register;
  blackhole gate must not docker-pause the shared central SQL Server
- mqtt: Sparkplug v3.0 STATE topic form; first-in-repo proto codegen noted
- omron: host hardcodes isIdempotent:false today (retry seam unshipped);
  v1 scopes UDTs to dotted-leaf access
- mtconnect: SecurityClassification.ViewOnly; factory ParseEnum<T> pattern
- program doc: both valid enum-serialization patterns; IRediscoverable is
  change-signal-gated; RTU P2 adds System.IO.Ports; label is host-side
2026-07-15 16:40:36 -04:00

293 lines
28 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# MQTT / Sparkplug B Driver — Research & Design
**Status:** Research / design proposal (not yet implemented)
**Author:** research sweep, 2026-07-15
**Scope:** A standard **Equipment-kind** driver (same shape as Modbus/S7/AbCip/TwinCAT/FOCAS/OpcUaClient) that ingests **plain MQTT** and **Sparkplug B** data into the OtOpcUa OPC UA / UNS address space.
This is a **subscribe-first** driver: the broker pushes data, we do not poll. Because OtOpcUa is a Unified-Namespace product, Sparkplug B alignment (the de-facto MQTT UNS payload standard) is the headline capability, with plain-MQTT/JSON as the broad-compatibility fallback.
---
## 1. Protocol summary + .NET library options
### 1.1 Plain MQTT
MQTT is a lightweight pub/sub transport: a client connects to a **broker**, subscribes to **topic filters** (with `+` single-level and `#` multi-level wildcards), and receives **retained** and live messages. Payloads are opaque byte arrays — by convention JSON, a scalar string/number, or a binary blob. There is no schema, no data-type metadata, and no built-in discovery: a consumer only knows a topic exists once a message arrives on it. "Last value" is available only if the publisher set the **retain** flag (the broker keeps the last retained message per topic and replays it to new subscribers).
Relevant knobs: QoS 0/1/2, `retain`, Last-Will-and-Testament (LWT), clean-session vs. persistent-session, TLS, username/password or client-cert auth, MQTT 3.1.1 vs 5.0.
### 1.2 Sparkplug B (Eclipse Tahu / Eclipse Sparkplug, spec v3.0)
Sparkplug B is an **open specification layered on MQTT** that adds the missing pieces for industrial data: a **mandated topic namespace**, a **protobuf payload schema**, **stateful session management**, and **auto-discovery via birth certificates**. It is the dominant "UNS over MQTT" standard.
**Topic namespace:** `spBv1.0/{group_id}/{message_type}/{edge_node_id}[/{device_id}]`
- `group_id` — logical grouping (e.g. a site/area).
- `edge_node_id` — an MQTT Edge-of-Network (EoN) node.
- `device_id` — an optional physical device attached to the edge node.
**Message types:**
| Type | Meaning | Direction |
|---|---|---|
| `NBIRTH` | Edge-node **birth certificate** — enumerates every node-level metric with name, alias, datatype, and initial value; carries `bdSeq` | edge → host |
| `DBIRTH` | Device **birth certificate** — enumerates a device's metrics (name/alias/datatype/value) | edge → host |
| `NDATA` / `DDATA` | Node/device **data** — metric changes, usually referenced by **alias** only (no name) | edge → host |
| `NDEATH` / `DDEATH` | Node/device **death** — LWT-driven; marks metrics STALE/uncertain | edge → host |
| `NCMD` / `DCMD` | **Command / write** to a node or device metric (e.g. setpoint, or `Node Control/Rebirth`) | host → edge |
| `STATE` | **Primary-host** application online/offline status (`ONLINE`/`OFFLINE`), retained, LWT-backed | host → edge |
**Key semantics the driver must honour:**
- **Birth-certificate discovery.** DBIRTH/NBIRTH is the *only* place metric **names + datatypes** appear. A consumer must cache the birth to interpret later data. This is exactly what makes Sparkplug **browsable** (§4).
- **Aliases.** After birth, NDATA/DDATA typically send only a numeric `alias` + value to save bandwidth. The consumer must maintain an **alias → (name, datatype)** map per edge-node/device, rebuilt on every (re)birth.
- **Sequence numbers.** Every Sparkplug payload carries a `seq` (0255, wraps) for gap detection; NBIRTH resets the sequence. `bdSeq` (birth/death sequence) in NBIRTH/NDEATH ties a death to its birth.
- **Rebirth.** If the host sees a gap, an unknown alias, or connects late (missed the birth), it issues an `NCMD` writing boolean `Node Control/Rebirth = true`; the edge node re-publishes NBIRTH + all DBIRTHs. This is the recovery primitive that lets a late-joining consumer recover full metadata.
- **Primary host / STATE.** A "primary host application" publishes a retained `STATE` message (with LWT set to `OFFLINE`, QoS 1, retain=true) so edge nodes know a trusted consumer is online. There **MUST be at most one** primary host client per host-id on a broker. OtOpcUa can consume as a non-primary application, or opt in as primary host (needed for guaranteed store-and-forward semantics on some edge devices).
- **Metric datatypes** (protobuf `DataType` enum): `Int8/16/32/64`, `UInt8/16/32/64`, `Float`, `Double`, `Boolean`, `String`, `DateTime`, `Text`, `UUID`, `Bytes`, `File`, `DataSet`, `Template`, plus array variants (`Int8Array`, …). Each metric = `{name, alias, timestamp, datatype, value, is_historical, is_transient, is_null}`.
Sources: [Eclipse Sparkplug spec (PDF, v2.2 — namespace/STATE rules carry into v3.0)](https://sparkplug.eclipse.org/specification/version/2.2/documents/sparkplug-specification-2.2.pdf) · [Eclipse Sparkplug normative statements](https://github.com/eclipse-sparkplug/sparkplug/blob/master/docs/normative_statements.md) · [Tahu `sparkplug_b.proto`](https://raw.githubusercontent.com/eclipse/tahu/master/sparkplug_b/sparkplug_b.proto) · [HiveMQ: Sparkplug session state](https://www.hivemq.com/blog/understanding-mqtt-topic-namespace-iiot/) · [Steve's Internet Guide: Sparkplug payloads/messages](http://www.steves-internet-guide.com/sparkplug-payloads-and-messages/) · [EMQX Sparkplug docs](https://docs.emqx.com/en/emqx/latest/data-integration/sparkplug.html)
### 1.3 .NET library options
| Library | License | Maturity | .NET 10 | Notes |
|---|---|---|---|---|
| **MQTTnet** | **MIT** | **High.** Now a **.NET Foundation** project hosted under **`dotnet/MQTTnet`**; the standard .NET MQTT client+server. Millions of NuGet downloads, active. v5.x current. | **Yes** — v5 explicitly added `dotnet10` target. | Client + broker, MQTT 3.1.1 & 5.0, TLS, WebSocket. This is the transport layer for **both** modes. |
| **SparkplugNet** | **MIT** | **Moderate.** Single-maintainer (SeppPenner), regular releases; latest **1.3.10 (2024-07-02)**. Supports Sparkplug **v3.0 / spBv1.0** (spAv1.0 obsolete). Provides `Application`, `Node`, `Device` base classes; the **Application** role receives N/DBIRTH + N/DDATA and can publish NCMD/DCMD. Lower adoption than MQTTnet. | **Targets net8.0/net9.0** today (plus "latest/LTS Core & Framework"). **No explicit net10 TFM yet** — will run on .NET 10 via net9.0 compat, but confirm at build time; may need an upstream TFM bump or a fork pin. | Wraps MQTTnet + `protobuf-net` internally. Depends on `MQTTnet >= 4.3.x` — watch for a **version clash** with a directly-referenced MQTTnet 5.x (see risk §6). |
| **MQTTnet + hand-decoded Tahu protobuf** | MIT (+ Tahu, EPL/Apache) | Build-it-yourself | Yes | Reference the Tahu `sparkplug_b.proto`, generate C# with `protobuf-net` or `Google.Protobuf`, decode payloads ourselves on top of an MQTTnet subscription. Maximum control (aliases, seq, rebirth logic exactly as we want), no third-party Sparkplug abstraction, but we own all the state-machine code. |
**Recommendation:** Use **MQTTnet (v5, MIT, .NET Foundation, net10-ready)** as the transport for both modes. For Sparkplug decoding, **start by evaluating SparkplugNet** to save the birth/alias/rebirth plumbing, but treat the **"MQTTnet + Tahu-proto hand-decode"** path as the fallback if SparkplugNet's net10 support or its internal MQTTnet-4.x pin proves awkward. The Sparkplug protobuf schema is small and stable, so hand-decoding is a bounded, well-understood effort.
Sources: [MQTTnet on NuGet (v5.2.0)](https://www.nuget.org/packages/MQTTnet/) · [dotnet/MQTTnet (GitHub)](https://github.com/dotnet/MQTTnet) · [MQTTnet — .NET Foundation](https://old.dotnetfoundation.org/projects/mqttnet) · [SparkplugNet on NuGet](https://www.nuget.org/packages/SparkplugNet) · [SparkplugNet (GitHub, SeppPenner)](https://github.com/SeppPenner/SparkplugNet)
---
## 2. Capability mapping
The driver implements the composable `IDriver` capability interfaces from `src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/`. Recommended set: `IDriver` + `ITagDiscovery` + `ISubscribable` + `IReadable` + `IHostConnectivityProbe` + `IRediscoverable`, and **optionally** `IWritable` (phase 2). Notably **not** a poller — unlike Modbus/S7 it does not use `PollGroupEngine`; it keeps a live broker connection and raises `OnDataChange` from the MQTT receive callback (closest existing analog: the native-push side of the OpcUaClient driver).
### 2.1 Subscribe (primary — `ISubscribable`)
The core of the driver. On `InitializeAsync`, connect to the broker (MQTTnet), set LWT/clean-session, and subscribe to the configured topic filters:
- **Plain mode:** subscribe to each authored tag's topic (or a shared wildcard) and to nothing else.
- **Sparkplug mode:** subscribe to `spBv1.0/{group_id}/#` (and `STATE/#` if primary host). Maintain per-edge-node/device **alias→metric** tables from N/DBIRTH; on N/DDATA, resolve each metric and raise `OnDataChange` with the authored tag's `FullReference`. On NDEATH/DDEATH, publish STALE/Bad-quality snapshots for that node's metrics. On a missed birth / unknown alias / seq gap, issue a **rebirth NCMD**.
`SubscribeAsync(fullReferences, publishingInterval, …)` maps requested `FullReference`s onto the live receive stream. MQTT/Sparkplug are inherently event-driven, so `publishingInterval` is advisory (used only for optional server-side coalescing/deadband, mirroring Modbus's `ShouldPublish`). The driver holds the broker connection for its whole lifetime; `OnDataChange` fires from the MQTT message handler. Emit an initial value from the retained message / last birth value on subscribe (OPC UA initial-data convention).
### 2.2 Discover (`ITagDiscovery`)
- **Sparkplug mode (rich):** DBIRTH/NBIRTH **enumerates** every metric with name + datatype → map directly onto Equipment + Tags. Natural UNS shape: `group_id` → Area/Line, `edge_node_id`/`device_id`**Equipment**, each **metric → Tag**. Because births arrive asynchronously after connect, use `RediscoverPolicy = UntilStable` (like FOCAS) — keep discovering until the observed birth set stops growing, and implement `IRediscoverable` so a *new* DBIRTH (new device joins, or a rebirth with changed metrics) triggers an address-space rebuild.
- **Plain mode (passive):** no schema. Two options: (a) **authored-only** — the operator declares tags (topic + JSON path) up front, `DiscoverAsync` returns exactly those (like Modbus's pre-declared `Tags`, `RediscoverPolicy = Once`); or (b) **observe-and-suggest** — subscribe `#` for a bounded window, build the observed topic tree, surface it in the browser (§4) for the operator to pick. Runtime discovery stays authored-only; wildcard observation is a **browse-time** concern, not an auto-provisioning one (avoids unbounded address spaces from chatty brokers).
### 2.3 Read (`IReadable`)
MQTT has no request/response read. `ReadAsync` returns the **last-value cache**: the last message seen on the topic (plain) or the last metric value from data/birth (Sparkplug). Seed it from the broker's **retained** message at connect (plain) and from birth values (Sparkplug). If no value has been observed yet, return `GoodNoData`/uncertain rather than an error. This satisfies OPC UA reads and HistoryRead-at-time without a live round-trip.
### 2.4 Write (`IWritable`) — **verdict: defer to phase 2, not in v1**
Writes are possible but semantically heavier and lower-priority for a UNS-ingest product:
- **Sparkplug:** publish an **NCMD/DCMD** to `spBv1.0/{group}/NCMD/{node}[/{device}]` with the target metric (by name/alias) + new value. Fire-and-forget at the MQTT layer — there is **no synchronous write ack**; success is only observable when the edge node echoes the change back in the next N/DDATA. This mirrors the Galaxy gateway's fire-and-forget write (Galaxy "can never surface a write failure"), so a write returns `Good` optimistically and the value self-corrects on the next data message. Respect `WriteIdempotent` — Sparkplug commands (rebirth, pulse) are frequently non-idempotent.
- **Plain:** publish to a configured command/write topic (often distinct from the read topic) with a formatted payload.
**Recommendation:** ship v1 **read/subscribe/discover only** (a genuinely useful UNS-ingest driver on its own), add `IWritable` (NCMD/DCMD + plain publish) in phase 2. Rationale: the write path has no ack model, needs careful idempotency/rebirth handling, and most UNS-ingest deployments are read-only consumers.
### 2.5 Data-type mapping (`DriverDataType`)
Map Sparkplug metric datatypes / inferred JSON types to `DriverDataType` (`src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/DriverDataType.cs`):
| Sparkplug `DataType` | `DriverDataType` | OPC UA |
|---|---|---|
| Int8, Int16 | Int16 | Int16 |
| Int32, UInt16 | Int32 / UInt16 | Int32 / UInt16 |
| Int64, UInt32 | Int64 / UInt32 | Int64 / UInt32 |
| UInt64 | UInt64 | UInt64 |
| Float | Float32 | Float |
| Double | Double64 → **Float64** | Double |
| Boolean | Boolean | Boolean |
| String, Text, UUID | String | String |
| DateTime | DateTime | DateTime |
| Bytes, File | (String/ByteString) | ByteString — phase 2 |
| DataSet, Template | **unsupported v1** | skip / raw-JSON string |
| `*Array` variants | `IsArray=true` + element type | ValueRank=1 |
Plain-MQTT/JSON: infer from the JSON token at the configured JSON path (number→Double/Int64, bool→Boolean, string→String), or honour an explicit per-tag `dataType` override (preferred — inference is brittle). `Int8`/`UInt8` widen to `Int16`/`UInt16` (no OPC UA byte-signed distinction needed). Complex `DataSet`/`Template` metrics are out of scope for v1 (flag + skip, or expose the raw JSON as a String tag).
---
## 3. TagConfig JSON shape
Two layers, mirroring every other driver: **driver options** (broker connection, one per driver instance) parsed by the driver factory, and **per-tag `TagConfig.FullName` + config** authored on the `/uns` Equipment → Tags tab.
### 3.1 Driver options (`MqttDriverOptions`, in `.Contracts`)
```jsonc
{
"host": "10.100.0.35",
"port": 8883,
"clientId": "otopcua-uns-1",
"useTls": true,
"allowUntrustedServerCertificate": false,
"caCertificatePath": null,
"username": "otopcua",
"password": "", // supply via env, never commit
"protocolVersion": "V500", // MQTT 3.1.1 | 5.0
"cleanSession": true,
"keepAliveSeconds": 30,
"reconnectMinBackoffSeconds": 1,
"reconnectMaxBackoffSeconds": 30,
"mode": "SparkplugB", // "Plain" | "SparkplugB"
// ---- Sparkplug-only ----
"sparkplug": {
"groupId": "Plant1", // subscribe spBv1.0/Plant1/#
"hostId": "otopcua-host-1", // STATE/{hostId} identity
"actAsPrimaryHost": false,
"requestRebirthOnGap": true,
"birthObservationWindowSeconds": 15 // discovery: how long to collect births
},
// ---- Plain-only ----
"plain": {
"topicPrefix": "factory/",
"defaultQos": 1
}
}
```
### 3.2 Per-tag config — Sparkplug
`TagConfig.FullName` is the driver-side full reference the router resolves. For Sparkplug the natural key is the fully-qualified metric path; store the full descriptor in the tag config so the driver can resolve by **name or alias** and survive alias reassignment across rebirths.
```jsonc
// TagConfig for one Sparkplug metric tag
{
"groupId": "Plant1",
"edgeNodeId": "Line3EdgeNode",
"deviceId": "Filler1", // omit/null for node-level metrics
"metricName": "Temperature/degC", // stable identity across rebirths
"dataType": "Float", // from DBIRTH; explicit for safety
"isHistorized": false
}
// FullName convention: "Plant1/Line3EdgeNode/Filler1:Temperature/degC"
```
### 3.3 Per-tag config — Plain MQTT
```jsonc
// TagConfig for one plain-MQTT tag
{
"topic": "factory/line3/oven/temp",
"payloadFormat": "Json", // "Json" | "Raw" | "Scalar"
"jsonPath": "$.value", // JSONPath into the payload (Json only)
"dataType": "Double", // explicit — inference is a fallback
"qos": 1,
"retainSeed": true // seed last-value from retained msg
}
// FullName convention: "factory/line3/oven/temp#$.value"
```
Resolution mirrors Modbus's `EquipmentTagRefResolver<TDef>`: the router hands the driver a `FullReference` (the raw `TagConfig` JSON or a stable string), the driver parses+caches it once, and matches inbound messages/metrics against it.
---
## 4. Browseability verdict + browse design
### Verdict: **BROWSEABLE — both modes, but discovery is passive and time-bounded.**
Unlike an OPC UA server (synchronous request/response `Browse`) or a Galaxy repository (queryable DB), MQTT/Sparkplug discovery is **observational**: you learn what exists only by *listening* for a while. The browse session therefore **connects to the broker and observes for a bounded window**, accumulating a tree, then answers `ExpandAsync` from that accumulated snapshot. This is the central design wrinkle.
- **Sparkplug (rich, structured):** subscribe `spBv1.0/{group}/#`, optionally publish a **rebirth NCMD** to force every edge node to re-announce immediately (turns a passive wait into a near-synchronous enumeration), and collect NBIRTH/DBIRTH. Births yield a clean, typed tree: **Group → EdgeNode → Device → Metric**, with datatypes — directly pickable, no guessing.
- **Plain (best-effort):** subscribe `#` (or a configured prefix) for the observation window and build the **observed topic tree**, splitting each topic on `/`. Leaves are topics that have carried a payload; the picker shows the last payload + inferred type as `AttributesAsync`. Coverage is only as good as broker traffic during the window (a topic silent during the window is invisible) — clearly a "suggestions" browse, not an authoritative enumeration.
### `IDriverBrowser` / `IBrowseSession` design
New project **`ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser`** (parallels `.OpcUaClient.Browser` / `.Galaxy.Browser`), implementing `Commons/Browsing/IDriverBrowser` + `IBrowseSession`, registered in `EndpointRouteBuilderExtensions` alongside the others:
```csharp
// EndpointRouteBuilderExtensions.cs (~line 49)
services.AddSingleton<IDriverBrowser, OpcUaClientDriverBrowser>();
services.AddSingleton<IDriverBrowser, GalaxyDriverBrowser>();
services.AddSingleton<IDriverBrowser, MqttDriverBrowser>(); // NEW
```
**`MqttDriverBrowser.OpenAsync(configJson, ct)`** — connect an MQTTnet client from the same `MqttDriverOptions` JSON the runtime driver consumes (separate client-id suffix so it never collides with the running driver's session), subscribe to the discovery filter, and **kick off the observation window immediately**. Returns an `MqttBrowseSession` whose internal tree fills in over the window (and keeps filling as long as the session lives). Because `IBrowseSession` methods are async and the session is registered in the `BrowseSessionRegistry` (TTL-reaped, like the OPC UA one), the "wait for observation" cost is naturally amortised across the user's clicks in the picker.
**`MqttBrowseSession`:**
- `RootAsync` — Sparkplug: the set of observed **groups** (or edge nodes under the configured group). Plain: top-level topic segments. Returns `BrowseNode { Kind = Folder, HasChildrenHint = true }`. If the window hasn't yielded anything yet, return what's accumulated so far (possibly empty) and let the user re-expand — or briefly await the first births. Optionally trigger a rebirth NCMD on first `RootAsync` in Sparkplug mode to populate fast.
- `ExpandAsync(nodeId)` — walk one level down the accumulated tree (EdgeNode→Device→Metric, or next topic segment). Metrics / leaf-topics are `Kind = Leaf`.
- `AttributesAsync(nodeId)` — Sparkplug: return the metric's `AttributeInfo { Name, DriverDataType, IsArray, SecurityClass }` from the cached birth. Plain: return the leaf topic's inferred type + last-seen payload snippet as a single synthetic attribute. (Sparkplug leaves are self-describing, so unlike OPC UA the side-panel is populated.)
- `DisposeAsync` — disconnect the MQTTnet client; best-effort (registry reaper may race a disconnect), same pattern as `OpcUaClientBrowseSession`.
**Picked node → TagConfig:** on commit, the picker projects the selected `BrowseNode`/`AttributeInfo` into the §3 TagConfig JSON. Sparkplug: `NodeId` encodes `group/node/device:metric`, split into the descriptor fields. Plain: `NodeId` is the topic path → `topic` + a default `jsonPath` (`$` or operator-edited) + inferred `dataType`.
**Addressing the passive/time-bounded wrinkle explicitly:**
1. Show a live "listening… N nodes/topics discovered" affordance in the picker so the operator understands coverage is accumulating, not instantaneous.
2. In Sparkplug mode, **actively force a rebirth** to convert the passive wait into a fast, near-complete enumeration.
3. Keep the browse session alive (registry TTL) so re-expanding later reflects newly observed topics without reconnecting.
4. Always allow **manual entry** of a topic/metric (the typed tag editor, §7) as an escape hatch when a tag is silent during the window.
---
## 5. Test-fixture strategy
Follow the existing driver-fixture pattern: a `docker-compose.yml` under `tests/Drivers/.../Docker/` with the `project: lmxopcua` label, deployed to the shared Docker host `10.100.0.35` via `lmxopcua-fix sync/up` (see `CLAUDE.md` Docker Workflow).
- **Broker:** **Eclipse Mosquitto** (`eclipse-mosquitto`, tiny, MIT/EPL) as the default fixture — carries both plain MQTT and Sparkplug B unchanged. Alternative **EMQX** (`emqx/emqx`, Apache-2.0, open-source CE up to 1000 conns) when a management dashboard or built-in Sparkplug tooling is wanted. HiveMQ CE is a third option.
- **Sparkplug edge-node simulator:** options, in rough order of convenience —
- **Bevywise MQTT/IoT Simulator** — purpose-built Sparkplug B realtime data simulation.
- **Eclipse Tahu** reference implementations (Java/Python) — author a small edge-node script that publishes NBIRTH/DBIRTH then periodic NDATA/DDATA; also exercises rebirth (subscribe NCMD `Node Control/Rebirth`). Most faithful to the spec.
- A **project-owned C# simulator** using the same MQTTnet + Tahu-proto path the driver uses (kills two birds: validates our encode/decode symmetrically).
- **Plain-MQTT fixture:** a tiny publisher container (`mosquitto_pub` loop, or a Python `paho-mqtt` script) emitting JSON on a handful of topics with `retain=true` so the last-value/read path and `#`-observation browse are testable.
- **Public test brokers** (for ad-hoc manual smoke only, never CI — unauthenticated, rate-limited, no privacy): `test.mosquitto.org`, `broker.hivemq.com`, `broker.emqx.io`. Do **not** put Sparkplug production-shaped data on public brokers.
- **Live-gate pattern:** an env-gated `*.IntegrationTests` suite (à la the HistorianGateway `LiveIntegration` category) that skips cleanly when `MQTT_FIXTURE_ENDPOINT` is unset, so the suite is macOS-offline-safe.
Sources: [Cedalo: Sparkplug B on Mosquitto](https://www.cedalo.com/blog/mqtt-sparkplug-mosquitto) · [EMQ: MQTT Sparkplug step-by-step](https://www.emqx.com/en/blog/mqtt-sparkplug-in-action-a-step-by-step-tutorial) · [Bevywise Sparkplug simulation](https://www.bevywise.com/blog/sparkplug-b-mqtt-simulation/) · [Ian Craggs: getting started with MQTT + Sparkplug](https://modelbasedtesting.co.uk/2022/01/22/getting-started-with-mqtt-and-sparkplug/)
---
## 6. Effort / risk / phasing
### Recommended order: **plain MQTT first, then Sparkplug B.**
Even though Sparkplug is the headline UNS capability, plain MQTT is the smaller, dependency-light slice that stands up the whole driver skeleton (project layout, `MqttDriverOptions`, factory/probe registration, `ISubscribable` receive loop, last-value `IReadable`, browser project, typed tag editor, fixtures). Sparkplug then layers the birth/alias/rebirth state machine + protobuf decode onto a proven connection+subscribe substrate. This also front-loads the reusable plumbing and de-risks the library decision (you can validate MQTTnet v5/net10 before committing to SparkplugNet).
| Phase | Scope | Rel. effort |
|---|---|---|
| **1 — Plain MQTT** | Projects (`Driver.Mqtt`, `.Contracts`, `.Browser`) + MQTTnet connect/TLS/auth/reconnect; `ISubscribable` (topic subscribe → `OnDataChange`); `IReadable` last-value (retained seed); authored-tag `ITagDiscovery`; `IHostConnectivityProbe`; `#`-observation browser; typed tag editor + validator; Mosquitto + JSON-publisher fixtures. | **M** |
| **2 — Sparkplug B ingest** | `spBv1.0/{group}/#` subscribe; Tahu-proto decode (SparkplugNet **or** hand-rolled); N/DBIRTH → discovery (`UntilStable` + `IRediscoverable`); alias tables; seq-gap detect + **rebirth NCMD**; N/DDEATH → STALE; STATE/primary-host option; birth-driven browser + `AttributesAsync`; Sparkplug datatype map; Tahu/Bevywise simulator fixture. | **L** |
| **3 — Write-through (optional)** | `IWritable`: Sparkplug NCMD/DCMD + plain publish; optimistic-Good + self-correct on echo; `WriteIdempotent` respect; write-topic config. | **M** |
### Top risks
1. **Sparkplug state-machine correctness (aliases + rebirth + seq).** The alias→metric mapping, sequence-gap detection, late-join rebirth, and death→STALE handling are the load-bearing logic and easy to get subtly wrong (e.g. an alias reused across a rebirth with a *different* metric silently mis-routes data to the wrong tag). Mitigation: bind tags by **stable metric name**, treat alias as a per-birth cache only; rebuild the alias table on every (re)birth; test explicitly against rebirth + missed-birth + gap scenarios with a controllable simulator. This is the single biggest correctness risk.
2. **Library / dependency friction (SparkplugNet ↔ MQTTnet version + net10).** SparkplugNet pins **MQTTnet 4.3.x** and has **no explicit net10 TFM** yet; a direct MQTTnet-5 reference (needed for the plain-mode transport + net10 support) can clash with SparkplugNet's transitive 4.x, and central-package-version pinning in this repo is already known to be fragile (see the Roslyn transitive-pinning memory). Mitigation: prototype the version matrix early; be ready to drop SparkplugNet and hand-decode the Tahu proto over a single MQTTnet-5 client (the proto is small/stable), which also removes the net10 TFM concern entirely.
Secondary risks: **passive-browse coverage gaps** (silent topics invisible in the observation window — mitigated by rebirth-forcing + manual entry, §4); **unbounded address space** from wildcard auto-provisioning (mitigated by authored-only runtime discovery); **write has no ack** (accepted, modelled as optimistic-Good self-correcting like Galaxy, §2.4); **broker security** (enforce TLS + real auth, never ship public-broker defaults).
---
## Appendix — project layout (mirrors existing drivers)
```
src/Drivers/
ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ # MqttDriver : IDriver, ITagDiscovery,
# ISubscribable, IReadable,
# IHostConnectivityProbe, IRediscoverable
# (+ IWritable in phase 3)
MqttDriver.cs, MqttDriverProbe.cs, MqttDriverFactoryExtensions.cs
Sparkplug/ (SparkplugDecoder, AliasTable, BirthCache, RebirthRequester)
ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/ # MqttDriverOptions, MqttTagDefinition,
# MqttEquipmentTagParser
ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ # MqttDriverBrowser, MqttBrowseSession
src/Server/.../AdminUI/
Uns/TagEditors/TagConfigEditorMap.cs # + ["Mqtt"] = MqttTagConfigEditor
Uns/TagEditors/TagConfigValidator.cs # + Mqtt validation
Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor
EndpointRouteBuilderExtensions.cs # + AddSingleton<IDriverBrowser, MqttDriverBrowser>
tests/Drivers/.../Docker/docker-compose.yml # mosquitto + sparkplug-sim + json-publisher
```
DriverType string: **`"Mqtt"`** (single canonical string across AdminUI page, probe, factory, editor map, browser — heed the Modbus `"ModbusTcp"` vs `"Modbus"` mismatch lesson; pick one and use it everywhere).