# MQTT / Sparkplug B driver — executable implementation design **Status:** design, build-ready. 2026-07-15. **Research input:** `docs/research/drivers/mqtt-sparkplug.md` (verdicts + protocol/library survey — this doc turns that into a build plan; it does not re-litigate the research). **Related:** `docs/plans/2026-07-15-universal-discovery-browser-design.md` (why MQTT needs a *bespoke* browser despite the universal one — §4), `docs/plans/2026-06-12-galaxy-standard-driver-design.md` (standard Equipment-kind driver shape + fire-and-forget write precedent), `docs/plans/2026-05-28-driver-browsers-design.md` (`IDriverBrowser`/`IBrowseSession` pattern reused by the observation browser). --- ## 1. Motivation + scope Build a standard **Equipment-kind** driver — same composable-`IDriver` shape as Modbus / S7 / AbCip / TwinCAT / FOCAS / OpcUaClient — that ingests **plain MQTT** and **Sparkplug B** into the OtOpcUa OPC UA / UNS address space. - OtOpcUa is a **Unified-Namespace product**, and **Sparkplug B is the de-facto UNS-over-MQTT payload standard** (mandated topic namespace + protobuf schema + birth-certificate discovery + stateful sessions). Sparkplug alignment is therefore the strategic headline; **plain MQTT/JSON** is the broad-compatibility fallback for brokers that carry raw JSON topics. - This is a **subscribe-first** driver. Unlike the polled drivers (Modbus/S7/AbCip/FOCAS with `PollGroupEngine`), it holds a **live broker connection** for its whole lifetime and raises `OnDataChange` from the MQTT receive callback. The closest existing analog is the native-push side of `OpcUaClientDriver`. - **In scope (this design):** connect/TLS/auth/reconnect; topic + Sparkplug subscribe → `OnDataChange`; last-value `IReadable`; authored-only `ITagDiscovery`; `IRediscoverable`; `IHostConnectivityProbe`; a **bespoke observation browser**; typed AdminUI tag editor + validator; Docker fixtures + env-gated live suite. - **Out of scope for v1:** write-through (`IWritable` — Sparkplug NCMD/DCMD + plain publish) is deferred to phase 3 (§3.4, §10); `DataSet`/`Template` Sparkplug metrics; `Bytes`/`File` metrics beyond a raw-String fallback. Canonical **DriverType string: `"Mqtt"`** — one string used identically across the AdminUI driver page, probe, factory, editor map, validator, and browser. (Heed the Modbus `"ModbusTcp"` vs `"Modbus"` mismatch lesson: a divergent string silently drops tags to the raw-JSON editor / faults resolution. Pick `"Mqtt"` and grep to enforce it.) --- ## 2. Project layout Three new projects under `src/Drivers/`, mirroring the OpcUaClient triad (`.Driver` + `.Contracts` + `.Browser`). All `net10.0`, `Nullable`+`ImplicitUsings` enabled, `TreatWarningsAsErrors=true` opted in per-csproj (mandatory for new projects — `Directory.Build.props` deliberately does **not** set it globally because legacy test projects would fail the build). ``` src/Drivers/ ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/ # options + tag DTOs + parser — zero transport deps MqttDriverOptions.cs # driver-config DTO (broker conn + mode + sparkplug/plain sub-objects) MqttMode.cs, MqttPayloadFormat.cs # enums (Plain|SparkplugB ; Json|Raw|Scalar) MqttTagDefinition.cs # parsed per-tag descriptor (plain OR sparkplug variant) MqttEquipmentTagParser.cs # TagConfig JSON → MqttTagDefinition (mirrors ModbusEquipmentTagParser) SparkplugDataType.cs # Sparkplug metric-datatype enum + → DriverDataType map ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ # runtime driver — references .Contracts + Core.Abstractions + Core MqttDriver.cs # IDriver, ITagDiscovery, ISubscribable, IReadable, # IHostConnectivityProbe, IRediscoverable (+ IWritable phase 3) MqttDriverProbe.cs # IDriverProbe — broker CONNECT handshake MqttDriverFactoryExtensions.cs # DriverTypeName="Mqtt"; Register(registry, loggerFactory) MqttConnection.cs # MQTTnet client wrapper: connect/TLS/auth/reconnect/backoff LastValueCache.cs # FullReference → last DataValueSnapshot (seeds IReadable) Sparkplug/ SparkplugCodec.cs # Tahu-proto decode (Payload/Metric); encode NCMD (phase 3) AliasTable.cs # per edge-node/device alias→(name,datatype); rebuilt each birth BirthCache.cs # NBIRTH/DBIRTH metric catalog (name/alias/datatype/last value) SequenceTracker.cs # seq (0-255 wrap) gap detection; bdSeq death→birth tie RebirthRequester.cs # publishes NCMD Node Control/Rebirth=true SparkplugTopic.cs # parse/format spBv1.0/{group}/{type}/{node}[/{device}] ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ # bespoke observation browser — references .Contracts + Commons(.Browsing) + MQTTnet MqttDriverBrowser.cs # IDriverBrowser (DriverType="Mqtt"); OpenAsync → MqttBrowseSession MqttBrowseSession.cs # IBrowseSession over an accumulating observed tree src/Server/.../AdminUI/ Uns/TagEditors/MqttTagConfigModel.cs # typed working model (FromJson/ToJson/Validate; preserves unknown keys) Uns/TagEditors/TagConfigEditorMap.cs # + ["Mqtt"] = MqttTagConfigEditor Uns/TagEditors/TagConfigValidator.cs # + ["Mqtt"] = j => MqttTagConfigModel.FromJson(j).Validate() Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor EndpointRouteBuilderExtensions.cs # + AddSingleton() src/Server/.../Host/Drivers/DriverFactoryBootstrap.cs # + MqttDriverFactoryExtensions.Register(registry, loggerFactory) # + TryAddEnumerable(IDriverProbe, MqttDriverProbe) tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ # unit — codec/alias/seq/parser (net10, macOS-safe) tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ # env-gated live suite (MQTT_FIXTURE_ENDPOINT) tests/Drivers/.../Docker/docker-compose.yml # mosquitto + sparkplug-sim + json-publisher (project=lmxopcua) ``` **Why the split `.Contracts` project:** it carries `MqttDriverOptions` + `MqttTagDefinition` + `MqttEquipmentTagParser` + the enums with **zero transport (MQTTnet) dependency**, exactly like `Driver.Modbus.Contracts` depends only on `Modbus.Addressing` + `Core.Abstractions`. This lets the AdminUI typed editor (and any tooling) reference the DTO shape/enums without dragging MQTTnet into AdminUI. The runtime `.Driver` and `.Browser` both reference `.Contracts` and add MQTTnet. ### 2.1 Libraries + the pinning caveat | Concern | Choice | Package | |---|---|---| | **MQTT transport (both modes)** | **MQTTnet v5** — MIT, **.NET Foundation** (`dotnet/MQTTnet`), the standard .NET MQTT client. v5 explicitly ships a `net10.0` TFM. Client + TLS + MQTT 3.1.1/5.0. **Note:** v5 dropped v4's `ManagedMqttClient` — reconnect is hand-rolled in P1 (§8). | `MQTTnet` (add `PackageVersion Include="MQTTnet" Version="5.x"` to `Directory.Packages.props`) | | **Sparkplug decode** | **Hand-rolled Tahu protobuf** (recommended) — generate C# from Tahu's `sparkplug_b.proto` with `Google.Protobuf` (`Grpc.Tools`/`protoc`), decode over a single MQTTnet-5 client. Small, stable schema; we own the alias/seq/rebirth state machine exactly as we want it. | `Google.Protobuf` + `Grpc.Tools` (build-time) | | **Alternative Sparkplug decode** | **SparkplugNet** — MIT, provides `Application`/`Node`/`Device` base classes, saves the birth/alias plumbing. **Rejected as primary** for two repo-specific reasons (below). | `SparkplugNet` (only if the risk clears) | **Pinning caveat (load-bearing — this is why hand-roll is recommended).** This repo uses **central package management** (`Directory.Packages.props`, `ManagePackageVersionsCentrally=true`) and deliberately **leaves `CentralPackageTransitivePinningEnabled` OFF** because turning it on breaks the Roslyn 5.0.0/4.12.0 split (see the "Transitive pinning breaks Roslyn build" memory + `Directory.Packages.props:103` comment). SparkplugNet **pins MQTTnet 4.3.x transitively and has no explicit `net10.0` TFM** yet. A direct MQTTnet-5 reference (needed for the plain-mode transport + net10) will **coexist in the graph with SparkplugNet's transitive 4.x**, and because transitive pinning is off we cannot cleanly force-align the two. Net: SparkplugNet risks a hard-to-resolve MQTTnet 4/5 diamond in exactly the pinning configuration this repo can't change. **Decision: hand-roll Tahu-proto decode over one MQTTnet-5 client.** It removes the diamond and the net10-TFM concern entirely, and the Sparkplug payload schema (`Payload { seq, timestamp, Metric[], body }`, `Metric { name, alias, timestamp, datatype, value, is_null, ... }`) is ~90 lines of `.proto` — a bounded, well-understood effort. Phase 1 (plain MQTT) references **only MQTTnet-5**, so we validate MQTTnet-v5-on-net10 before ever touching Sparkplug. (If a future need for SparkplugNet arises, the phase-2 codec seam `SparkplugCodec` is the single swap point.) - [ ] **Re-verify the external-library claims at Wave-2 start (before P1 coding):** confirm the exact **MQTTnet v5 version + `net10.0` TFM** on NuGet, and re-check the **SparkplugNet MQTTnet-4.x transitive-pin** claim — it is directionally right from the research but was unverifiable offline. The hand-roll decision does **not** hinge on that pin being exact (it stands on the net10-TFM concern + owning the alias/seq/rebirth state machine regardless), so this is a confirm-the-record check, not a decision gate. **Vendor the proto:** commit Tahu `sparkplug_b.proto` into `Driver.Mqtt.Contracts/Protos/` (pinned to a known Tahu commit, with a provenance comment) and generate at build via `Grpc.Tools`. Don't fetch at build time. **Existing protobuf tooling, honestly:** `Google.Protobuf` is **already centrally pinned** (`Directory.Packages.props:31`, 3.34.1) and consumed by `Driver.Galaxy` — but the Galaxy proto types ship **pre-generated** inside the `ZB.MOM.WW.MxGateway.*` NuGets; there is **no `.proto` file or `Grpc.Tools` protoc codegen anywhere in this repo today**. So `Grpc.Tools` is a new (build-time-only, `PrivateAssets=all`) package and this is the repo's first in-repo proto codegen step. If that step proves objectionable, the fallback is checking in the generated C# (the schema is small and stable) with the same provenance comment. --- ## 3. Capability mapping `MqttDriver : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable` (+ `IWritable` in phase 3). **Not** `IHistoryProvider` (historization is the server-side `isHistorized` path, unchanged). **Not** a poller — no `PollGroupEngine`. ### 3.0 Lifecycle (`IDriver`) - `InitializeAsync(configJson, ct)` — deserialize `MqttDriverOptions` (shared `JsonSerializerOptions` with `JsonStringEnumConverter`, §6); build `MqttConnection`; connect to broker under a bounded connect deadline (§8); set LWT/clean-session; subscribe to the mode-appropriate filter (§3.1). Publish `STATE` birth if `actAsPrimaryHost` (Sparkplug). - `ReinitializeAsync` — apply a config delta in place: if only tag membership changed, re-subscribe; if broker connection changed, tear down + reconnect. Never crash the process (marks Faulted → nodes go Bad, per driver-stability Tier A/B). - `ShutdownAsync` — publish `STATE OFFLINE` (if primary host), disconnect the MQTTnet client, dispose. - `GetHealth` — Connected/Reconnecting/Faulted + last-message-age; drives ServiceLevel + status dashboard. - `GetMemoryFootprint`/`FlushOptionalCachesAsync` — footprint = last-value cache + birth caches + alias tables. The runtime driver has **no optional caches to flush**: birth/alias tables are required-for-correctness (flushing them mis-routes aliased DDATA — `IDriver.FlushOptionalCachesAsync` forbids flushing correctness state) and the last-value cache is what `IReadable` serves. Flush is a no-op; all three structures are bounded by the authored tag set + observed births. (The observation browser's accumulating tree lives in the AdminUI-side `MqttBrowseSession`, not in the runtime driver.) ### 3.1 Subscribe (`ISubscribable` — primary) - **Plain mode:** subscribe to each authored tag's `topic` at its `qos` (dedupe identical topics; coalesce shared wildcards where the operator authored one). On message: match topic → authored tag(s), extract value at `jsonPath` (or raw/scalar), raise `OnDataChange(handle, FullReference, snapshot)`. Seed initial value from the broker's **retained** message replayed on subscribe (OPC UA initial-data convention). - **Sparkplug mode:** subscribe `spBv1.0/{groupId}/#` (+ the host-state topic when acting as primary host — **`spBv1.0/STATE/{hostId}` with a JSON `{online, timestamp}` payload in spec v3.0**; pre-3.0 peers used top-level `STATE/{hostId}` with `ONLINE`/`OFFLINE` text — target the v3.0 form, tolerate the legacy one on receive). Maintain a **`BirthCache` + `AliasTable` per (edgeNode, device)** from NBIRTH/DBIRTH. On NDATA/DDATA: resolve each metric by **alias → (name, datatype)** via the alias table, map to the authored tag's `FullReference` **by stable metric name**, raise `OnDataChange`. On NDEATH/DDEATH: emit **STALE / Bad-quality** snapshots for that node's/device's metrics. On missed birth / unknown alias / seq gap: issue a **rebirth NCMD** (§3.6). **Spec version:** implement from the **Sparkplug v3.0 spec text** — the research report cites v2.2, which is superseded on exactly the STATE topic/payload point above (this doc already carries the correct v3.0 form). - `publishingInterval` is **advisory** (MQTT/Sparkplug are event-driven). Use it only for optional server-side coalescing/deadband, mirroring Modbus's `ShouldPublish`. `SubscribeAsync` returns an `ISubscriptionHandle` whose `DiagnosticId` names the mode+filter; `OnDataChange` fires from the MQTT receive handler for the whole driver lifetime. `SubscribeAsync` maps the requested `FullReference`s onto the live receive stream: it registers them in a `FullReference → MqttTagDefinition` resolver (parsed once, cached — reuse the shared `EquipmentTagRefResolver` from `Core.Abstractions`, exactly as Modbus does), so an inbound message/metric routes to the right node. ### 3.2 Read (`IReadable`) MQTT has no request/response read. `ReadAsync(fullReferences)` returns the **last-value cache** (`LastValueCache`): last message seen on the topic (plain) or last metric value from data/birth (Sparkplug). Seed from the retained message at connect (plain) and from birth values (Sparkplug). If nothing has been observed yet, return a `GoodNoData`/uncertain snapshot **per reference** (never throw the batch — per `IReadable` contract, per-reference status is the snapshot's `StatusCode`). This satisfies OPC UA reads + HistoryRead-AtTime without a live round-trip. ### 3.3 Discover (`ITagDiscovery`) — authored-only **`DiscoverAsync` replays authored tags only, both modes** — it does **not** auto-provision from wildcard/birth observation. Rationale (from research §2.2): a chatty broker or a large Sparkplug plant would produce an unbounded auto-provisioned address space. Wildcard/birth observation is a **browse-time** concern (§4), not a runtime-provisioning one. - The operator authors tags (topic+jsonPath, or group/node/device/metric) via the typed editor / observation browser. `DiscoverAsync` streams exactly those authored tags into the `IAddressSpaceBuilder`, each as a `Variable(browseName, displayName, DriverAttributeInfo{ FullName, DriverDataType, ... })`. - **`RediscoverPolicy`:** - **Plain:** `Once` (authored set is static per published config — like Modbus). - **Sparkplug:** `UntilStable` (metric *datatypes* fill in asynchronously as births arrive after connect; keep re-streaming until the authored tags' resolved datatypes stop changing — like FOCAS's FixedTree). This lets a tag authored before its first birth pick up its true datatype once the birth lands. - **`SupportsOnlineDiscovery => false`** (the default-member from the universal-browser design). This is deliberate and **decoupled from browseability** (§4): runtime discovery is authored-only, so the universal Discover-backed browser would only replay authored tags — useless for discovering *new* topics/metrics. Leaving the flag false keeps MQTT off the universal fallback; the bespoke observation browser (§4) is what backs the picker. ### 3.4 `IRediscoverable` Sparkplug fires `OnRediscoveryNeeded` when a **new DBIRTH** (a device joins) or a **rebirth with a changed metric set** is observed — the Core re-runs `DiscoverAsync` + diffs. `ScopeHint` = the edge-node/device folder path so the rebuild is surgical. Plain mode never fires it (no backend change signal — a config republish is the only tag-set change), so plain-only drivers effectively no-op this interface. ### 3.5 Data-type mapping Sparkplug metric datatype (or inferred/authored plain type) → `DriverDataType` (`Boolean, Int16, Int32, Int64, UInt16, UInt32, UInt64, Float32, Float64, String, DateTime, Reference`): | Sparkplug `DataType` | `DriverDataType` | OPC UA | |---|---|---| | Int8, Int16 | `Int16` (Int8 widens) | Int16 | | Int32 | `Int32` | Int32 | | UInt8, UInt16 | `UInt16` (UInt8 widens) | UInt16 | | UInt32 | `UInt32` | UInt32 | | Int64 | `Int64` | Int64 | | UInt64 | `UInt64` | UInt64 | | Float | `Float32` | Float | | Double | `Float64` | Double | | Boolean | `Boolean` | Boolean | | String, Text, UUID | `String` | String | | DateTime | `DateTime` | DateTime | | Bytes, File | `String` (base64/raw fallback, v1) | String | | DataSet, Template | **unsupported v1** — skip + warn, or expose raw-JSON as `String` | — | | `*Array` variants | element type + `IsArray=true`, `ArrayDim` from birth | ValueRank=1 | Live map in `SparkplugDataType.ToDriverDataType()` (`.Contracts`). **Plain mode:** honour an explicit per-tag `dataType` (preferred — inference is brittle); fall back to inferring from the JSON token at `jsonPath` (number→`Float64`/`Int64`, bool→`Boolean`, string→`String`). Int8/UInt8 widen (no OPC UA signed-byte distinction needed). ### 3.6 Sparkplug state-machine correctness (the #1 risk) The alias→metric map, seq-gap detection, late-join rebirth, and death→STALE are load-bearing and easy to get subtly wrong. **Invariants the implementation must hold:** 1. **Bind tags by stable metric NAME, never by alias.** The authored `TagConfig` stores `metricName`; the alias is a **per-birth cache only**. An alias can be reused across a rebirth for a *different* metric — binding by alias silently mis-routes data to the wrong tag. 2. **Rebuild the alias table on every (re)birth.** NBIRTH/DBIRTH replaces the `(edgeNode,device)` `AliasTable` wholesale; never merge. 3. **Track `seq` (0–255, wraps) per edge-node.** A gap (or an unknown alias, or a data message before any birth) ⇒ request rebirth (`RebirthRequester` publishes `NCMD` writing `Node Control/Rebirth = true`), gated on `requestRebirthOnGap`. `bdSeq` ties an NDEATH to its NBIRTH. 4. **Death → STALE.** NDEATH/DDEATH (LWT-driven) emits Bad/uncertain-quality snapshots for all affected metrics; the next birth restores Good. 5. **Late join.** On connect the driver may have missed births — proactively request rebirth (or wait for the retained STATE + natural rebirth) so metadata is recovered. (This runtime-driver late-join rebirth is automatic and correct — the driver *owns* its subscription; it is the same NCMD primitive the browser exposes only as the explicit operator "Request rebirth" action, §4.1.) Unit-test these explicitly against a controllable simulator: rebirth, missed-birth, seq-gap, alias-reuse-across-rebirth, death→stale→rebirth. ### 3.7 Write (`IWritable`) — **verdict: NOT in v1; phase 3, fire-and-forget optimistic-Good** Writes are possible but deferred (research §2.4). When shipped in phase 3: - **Sparkplug:** publish `NCMD`/`DCMD` to `spBv1.0/{group}/NCMD/{node}[/{device}]` with the target metric (by name+alias) + value. **No synchronous write ack** — success is only observable when the edge echoes the change in the next N/DDATA. This mirrors the **Galaxy gateway's fire-and-forget write** ("Galaxy can never surface a write failure"): return `Good` optimistically; the value self-corrects on the next data message. Respect `WriteIdempotent` — Sparkplug commands (rebirth, pulse) are frequently **non-idempotent**, so default retry off. - **Plain:** publish to a configured command/write topic with a formatted payload. v1 ships **read/subscribe/discover only** — a genuinely useful UNS-ingest driver. `MqttDriver` does **not** implement `IWritable` in v1 (the interface is composable; its absence means nodes materialize read-only). --- ## 4. Bespoke observation browser ### Decision — reconciling with the universal Discover-backed browser The universal browser (`docs/plans/2026-07-15-universal-discovery-browser-design.md`) captures whatever `DiscoverAsync` streams and re-presents it, gated on `SupportsOnlineDiscovery`. **MQTT/Sparkplug is the one roadmap driver that needs a bespoke browser despite the universal one**, because: - Runtime `DiscoverAsync` is **authored-only** (§3.3). A universal Discover-backed browser would replay only tags the operator already authored — it would **not discover new topics/metrics**, which is the entire point of a picker. - MQTT/Sparkplug discovery is **observational**: you learn what exists only by *listening* for a bounded window. That's a fundamentally different session shape than "connect → one-shot enumerate → disconnect" — the tree **fills in over time**, and Sparkplug can be *accelerated on demand* by an operator-requested rebirth NCMD (§4.1 — explicitly not automatic). **Considered + rejected:** feeding the Sparkplug `BirthCache` into `DiscoverAsync` device-enumeration to plug into universal. Rejected because it would make runtime discovery auto-provision from births (the unbounded-address-space failure §3.3 explicitly avoids) and still wouldn't cover plain MQTT. The birth cache stays a browse-time + subscribe-time structure, not a discovery source. **Decision:** keep `SupportsOnlineDiscovery=false` (authored-only runtime discovery) **and** build a bespoke `IDriverBrowser` (`MqttDriverBrowser`) with an **observation-window `IBrowseSession`** (`MqttBrowseSession`). Registering the bespoke browser for `DriverType="Mqtt"` **overrides the universal fallback** for MQTT (the universal browser resolves bespoke-first — `BrowserSessionService` §3.1 of that doc). ### 4.1 `MqttDriverBrowser : IDriverBrowser` `DriverType => "Mqtt"`. `OpenAsync(configJson, ct)`: 1. Parse the **same `MqttDriverOptions` JSON** the runtime driver consumes (shared `JsonSerializerOptions`), but connect with a **distinct client-id suffix** (`{clientId}-browse-{guid8}`) so the browse session never collides with the running driver's MQTT session / clean-session state. 2. Connect an MQTTnet client under a bounded connect budget (clamp like OpcUaClient's `PerEndpointConnectTimeout`, 5–30 s). 3. Subscribe to the discovery filter: `spBv1.0/{groupId}/#` (Sparkplug) or `#`/`{topicPrefix}#` (plain). 4. **Kick off the observation window immediately — passively.** The session only *listens*; opening it publishes **nothing**. This keeps browse read-only by default, matching the house browse pattern (driver-browsers design §6: "browse is read-only and doesn't mutate config or driver state"), and avoids rebirth-storming every edge node in the group whenever an operator merely opens a picker — on a large plant an auto-rebirth on session open would stampede the whole group's edge nodes into re-publishing full births. 5. Return an `MqttBrowseSession` whose internal tree **fills in over the window and keeps filling** while the session lives. Because `IBrowseSession` methods are async and the session lands in the AdminUI `BrowseSessionRegistry` (TTL-reaped, per-call 20 s timeout), the observation cost is amortised across the user's clicks in the picker. **Explicit "Request rebirth" — the one operator-initiated write (Sparkplug mode only).** The picker carries a **"Request rebirth"** button, enabled in Sparkplug mode and **scoped to the currently selected group or edge node**. Clicking it invokes `MqttBrowseSession.RequestRebirthAsync(scope)` — an MQTT-specific session method *beyond* the base `IBrowseSession` surface, dispatched by the picker through `BrowserSessionService` for MQTT sessions — which publishes the rebirth **NCMD** (`Node Control/Rebirth = true`, via the same `RebirthRequester` encode path) to the selected edge node, or to each *observed* edge node under the selected group. That converts the passive wait into a near-synchronous enumeration precisely when the operator asks for it. This is the design's single deliberate deviation from browse-is-read-only: it is operator-initiated, scoped, gated by the same `DriverOperator` policy that gates the picker, and logged (Info, with the target scope) — it is **never** fired automatically by `OpenAsync`, `RootAsync`, or expansion. ### 4.2 `MqttBrowseSession : IBrowseSession` Serves from an **accumulating observed tree** (thread-safe; the MQTT receive handler mutates it, browse calls read it). Refreshes `LastUsedUtc` on each call. - **`RootAsync`** — Sparkplug: observed **groups** (or edge nodes under the configured group). Plain: top-level topic segments. `BrowseNode { Kind=Folder, HasChildrenHint=true }`. If the window hasn't yielded yet, return what's accumulated (possibly empty) and let the user re-expand — or, in Sparkplug mode, click the explicit **"Request rebirth"** button (§4.1) to populate fast. `RootAsync` itself never publishes; browse calls are pure reads of the accumulated tree. - **`ExpandAsync(nodeId)`** — one level down the accumulated tree. Sparkplug: **Group → EdgeNode → Device → Metric**. Plain: next topic segment (split on `/`). Metrics / leaf-topics are `Kind=Leaf`. - **`AttributesAsync(nodeId)`** — Sparkplug: the metric's `AttributeInfo { Name, DriverDataType (from birth), IsArray, SecurityClass }` — **self-describing**, so the picker side-panel is populated (unlike the OPC UA browser which returns empty). Plain: a single synthetic attribute carrying the leaf topic's **inferred type + last-seen payload snippet**. - **`RequestRebirthAsync(scope)`** — Sparkplug only, *beyond* the base `IBrowseSession` surface (§4.1): the single session member that publishes, and only ever on explicit operator click. - **`DisposeAsync`** — disconnect the MQTTnet client, best-effort (registry reaper may race), same pattern as `OpcUaClientBrowseSession`. ### 4.3 Picked node → TagConfig On commit the picker projects the selected `BrowseNode`/`AttributeInfo` into the §5 TagConfig JSON: - **Sparkplug:** the `NodeId` encodes `group/node/device:metric` → split into `groupId`/`edgeNodeId`/`deviceId`/`metricName`; `dataType` from `AttributeInfo`. - **Plain:** the `NodeId` is the topic path → `topic` + a default `jsonPath` (`$`, operator-editable) + inferred `dataType`. ### 4.4 Addressing the passive/time-bounded wrinkle explicitly 1. Picker shows a live **"listening… N nodes/topics discovered"** affordance so the operator understands coverage is accumulating, not instantaneous. 2. Sparkplug mode offers the explicit **"Request rebirth"** button (§4.1) — operator-initiated, scoped to the selected group/node — to convert the passive wait into a fast near-complete enumeration on demand. The window itself stays passive; if the operator doesn't click, coverage accumulates from natural traffic only. 3. The browse session stays alive (registry TTL) so re-expanding later reflects newly observed topics without reconnecting. 4. **Manual entry** of a topic/metric via the typed editor (§6) is always available as an escape hatch for a tag that's silent during the window. ### 4.5 Registration ```csharp // EndpointRouteBuilderExtensions.cs — alongside the existing two (near line 49-50) services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); // NEW — overrides universal fallback for "Mqtt" ``` --- ## 5. TagConfig + driver-config JSON Two layers, mirroring every other driver: **driver options** (broker connection, one per driver instance, parsed by the factory) and **per-tag config** (`TagConfig` JSON authored on the `/uns` Equipment → Tags tab; the driver's `FullName` is the driver-side reference the router resolves). ### 5.1 Driver options — `MqttDriverOptions` (`.Contracts`) ```jsonc { "host": "10.100.0.35", "port": 8883, "clientId": "otopcua-uns-1", "useTls": true, "allowUntrustedServerCertificate": false, // dev/on-prem only "caCertificatePath": null, // PEM CA pin; null ⇒ OS trust store "username": "otopcua", "password": "", // supply via env, NEVER commit "protocolVersion": "V500", // "V311" | "V500" "cleanSession": true, "keepAliveSeconds": 30, "connectTimeoutSeconds": 15, // bounded connect deadline (§8) "reconnectMinBackoffSeconds": 1, "reconnectMaxBackoffSeconds": 30, "mode": "SparkplugB", // "Plain" | "SparkplugB" // ---- Sparkplug-only (null in Plain mode) ---- "sparkplug": { "groupId": "Plant1", // subscribe spBv1.0/Plant1/# "hostId": "otopcua-host-1", // spBv1.0/STATE/{hostId} identity (v3.0) "actAsPrimaryHost": false, // at most ONE primary host per host-id per broker "requestRebirthOnGap": true, "birthObservationWindowSeconds": 15 // browse/discovery: how long to collect births }, // ---- Plain-only (null in SparkplugB mode) ---- "plain": { "topicPrefix": "factory/", "defaultQos": 1 } } ``` ### 5.2 Per-tag — Sparkplug ```jsonc { "FullName": "Plant1/Line3EdgeNode/Filler1:Temperature/degC", // PascalCase key (composer/walker contract) "groupId": "Plant1", "edgeNodeId": "Line3EdgeNode", "deviceId": "Filler1", // omit/null for node-level metrics "metricName": "Temperature/degC", // STABLE identity across rebirths — bind by this, not alias "dataType": "Float" // from DBIRTH; explicit for safety } ``` `FullName` convention: `"{group}/{node}[/{device}]:{metric}"`. The driver parses `FullName` (or the descriptor fields) once, caches it, and resolves inbound metrics by **`(group,node,device,metricName)`**. ### 5.3 Per-tag — Plain MQTT ```jsonc { "FullName": "factory/line3/oven/temp#$.value", // topic#jsonPath — stable per-tag key "topic": "factory/line3/oven/temp", "payloadFormat": "Json", // "Json" | "Raw" | "Scalar" "jsonPath": "$.value", // JSONPath into payload (Json only) "dataType": "Double", // explicit — inference is the fallback "qos": 1, "retainSeed": true // seed last-value from retained msg on subscribe } ``` `FullName` convention: `"{topic}#{jsonPath}"` (Json) or just `"{topic}"` (Raw/Scalar). `MqttEquipmentTagParser.TryParse(reference, out def)` follows `ModbusEquipmentTagParser`: a leading `{` marks a TagConfig blob; parse it, use **strict enum reads** (`TagConfigJson.TryReadEnumStrict`) so a typo'd `payloadFormat`/`dataType` rejects the tag (→ `BadNodeIdUnknown`) rather than silently defaulting to a wrong type; the def's `Name` = the reference string so published values key the forward router. Add an `Inspect(reference)` deploy-time warning pass (like Modbus) for present-but-invalid enums / unparseable blobs. **Historization:** the server-side `isHistorized` / `historianTagname` keys are owned by the TagModal-merge seam (`TagHistorizeConfig`) and survive load→save of the typed model as preserved unknown keys — the driver does not model them (identical to `OpcUaClientTagConfigModel`). --- ## 6. Typed editor + validator ### 6.1 `MqttTagConfigModel` (`.AdminUI/Uns/TagEditors/`) Thin typed working model over a preserved `JsonObject` key bag, per the `OpcUaClientTagConfigModel` template: `FromJson(json)` / `ToJson()` / `Validate()`, **preserving unknown keys** across a load→save (so history keys + fields this editor doesn't expose survive). It carries **both** mode shapes — a `Mode` field (Plain|SparkplugB, defaulted from which fields are present) selects which sub-shape the editor renders + validates. - `FromJson` reads the descriptor fields for the active mode plus the always-present `FullName` (PascalCase, case-sensitive — the composer/artifact-decoder/walker read it via case-sensitive `TryGetProperty("FullName", …)`). - `ToJson` writes `FullName` PascalCase and re-derives it from the descriptor fields (`{group}/{node}[/{device}]:{metric}` or `{topic}#{jsonPath}`) so a hand-edited descriptor stays consistent with the routed key. - `Validate()`: - **Sparkplug:** `groupId`, `edgeNodeId`, `metricName` required; `deviceId` optional; `dataType` must be a known Sparkplug type. - **Plain:** `topic` required and non-wildcard (a subscription topic for a *tag* must be concrete — reject `+`/`#`); `payloadFormat=Json` ⇒ `jsonPath` required; `dataType` a known `DriverDataType`. ### 6.2 `MqttTagConfigEditor.razor` Copy the Modbus editor template (`Components/Shared/Uns/TagEditors/`): a razor shell over the pure model, mode-switch at top, per-mode field group below, client-side validation via `Validate()`. Register: ```csharp // TagConfigEditorMap.cs ["Mqtt"] = typeof(Components.Shared.Uns.TagEditors.MqttTagConfigEditor), // TagConfigValidator.cs ["Mqtt"] = j => MqttTagConfigModel.FromJson(j).Validate(), ``` ### 6.3 The `JsonStringEnumConverter` enum-serialization trap **Every** JSON seam that (de)serializes `MqttDriverOptions` / tag configs — factory, probe, browser, **and the AdminUI driver-config page + probe DTO** — MUST use a `JsonSerializerOptions` that includes `new JsonStringEnumConverter()` (plus `PropertyNameCaseInsensitive=true`, `UnmappedMemberHandling=Skip`). The repo's SYSTEMIC bug (memory: "Driver enum-serialization bug"): AdminUI pages that serialize enums **numerically** while factory DTOs are **string-typed** fault the driver on any enum field. `MqttMode`, `MqttPayloadFormat`, `protocolVersion` are all enums — the AdminUI Mqtt driver page and its probe must serialize them as **strings** (mirror the OpcUaClient page). `JsonStringEnumConverter` also accepts numeric ordinals, so existing numeric configs still load. Keep one shared `JsonSerializerOptions` definition referenced by factory + probe + browser (as OpcUaClient does). --- ## 7. Factory + registration `MqttDriverFactoryExtensions` (mirror `OpcUaClientDriverFactoryExtensions`): ```csharp public static class MqttDriverFactoryExtensions { public const string DriverTypeName = "Mqtt"; private static readonly JsonSerializerOptions JsonOptions = new() { PropertyNameCaseInsensitive = true, UnmappedMemberHandling = JsonUnmappedMemberHandling.Skip, Converters = { new JsonStringEnumConverter() }, }; public static void Register(DriverFactoryRegistry registry, ILoggerFactory? loggerFactory = null) { ArgumentNullException.ThrowIfNull(registry); registry.Register(DriverTypeName, (id, json) => CreateInstance(id, json, loggerFactory)); } public static MqttDriver CreateInstance(string driverInstanceId, string driverConfigJson, ILoggerFactory? lf = null) { var options = JsonSerializer.Deserialize(driverConfigJson, JsonOptions) ?? throw new InvalidOperationException($"Mqtt driver config for '{driverInstanceId}' deserialised to null"); return new MqttDriver(options, driverInstanceId, lf?.CreateLogger()); } } ``` Wire in the two host bootstrap sites: ```csharp // Host/Drivers/DriverFactoryBootstrap.cs — factory registration (near line 133) Driver.Mqtt.MqttDriverFactoryExtensions.Register(registry, loggerFactory); // … and probe registration in AddOtOpcUaDriverProbes (near line 114) services.TryAddEnumerable(ServiceDescriptor.Singleton()); ``` **`MqttDriverProbe : IDriverProbe`** (`DriverType => "Mqtt"`): parse the config with the **shared** `JsonSerializerOptions`, open an MQTTnet CONNECT to `host:port` under the timeout, return green + latency on CONNACK-accepted, or a targeted error (connection refused / TLS failure / auth failure / timeout). Analogous to `ModbusDriverProbe`'s TCP+FC03 handshake — CONNACK is the MQTT "device is answering" proof. **Probes are admin-pinned** (the `admin-operations` singleton dispatches by DriverType) — the `TryAddEnumerable` line must be on admin nodes (it is, via `AddOtOpcUaDriverProbes` in `Program.cs`'s `hasAdmin` block). --- ## 8. Resilience / timeout / security - **Bounded connect.** `InitializeAsync` connects under `connectTimeoutSeconds` (default 15) via a linked CTS — **no unbounded wait** on a dead broker (heed the S7/FOCAS read-timeout lessons: an async network call that ignores its deadline wedges the whole driver). - **Reconnect with backoff — hand-rolled, committed for P1.** MQTTnet v5 **dropped** the v4 `ManagedMqttClient`, so there is no library-managed reconnect to lean on: `MqttConnection` owns a hand-rolled reconnect loop. Requirements: exponential backoff `reconnectMinBackoffSeconds → reconnectMaxBackoffSeconds`; **re-subscribe on every reconnect** (never assume broker-held subscription state — with `cleanSession=true` subscriptions don't survive the session, and even with a persistent session treat re-subscribe as idempotent insurance); and session-state handling (honour `cleanSession`/session-expiry semantics; re-establish the Sparkplug STATE birth when acting as primary host). On disconnect: `GetHealth` → Reconnecting; on reconnect: re-subscribe, and in Sparkplug mode **request rebirth** to recover metadata (late-join, §3.6). Nodes go Bad-quality while disconnected via the standard fan-out. - **Subscribe deadline.** `SubscribeAsync` completes under a bounded deadline; SUBACK failure surfaces as a per-reference Bad, not a hang. - **No poll loop** — the receive handler must be non-blocking (offload heavy decode off the MQTT callback thread; never block the MQTTnet dispatcher). - **Broker security — enforce, never ship public-broker defaults.** Default `useTls=true`; real auth (username/password or client-cert). `allowUntrustedServerCertificate` + `caCertificatePath` mirror the ServerHistorian TLS knobs (dev/on-prem escape hatch, off by default). **Never** put Sparkplug production-shaped data on public brokers (`test.mosquitto.org` etc. — manual smoke only). Credentials come from env (`password` blank in committed JSON), same posture as `ServerHistorian__ApiKey`. - **Primary-host guard.** At most **one** primary-host client per `hostId` per broker (Sparkplug MUST). `actAsPrimaryHost=true` on two OtOpcUa instances sharing a `hostId` (e.g. a redundancy pair) is a misconfiguration — the redundancy pair should use **distinct** host-ids (mirrors the distinct `MxAccess.ClientName` redundancy rule). --- ## 9. Test fixtures Follow the driver-fixture pattern: `tests/Drivers/.../Docker/docker-compose.yml` with the `project: lmxopcua` label, deployed to `10.100.0.35` via `lmxopcua-fix sync/up` (`CLAUDE.md` Docker Workflow). Repo files are source-of-truth; `/opt/otopcua-mqtt/` is the mirrored deployment. - **Broker:** **Eclipse Mosquitto** (`eclipse-mosquitto`, tiny, carries both plain + Sparkplug unchanged) as the default fixture; **EMQX** (`emqx/emqx`, Apache-2.0 CE) as the alternative when a dashboard / built-in Sparkplug tooling helps. Configure with **auth + TLS** so the fixtures exercise the real security path (not anonymous). - **Sparkplug edge-node simulator:** prefer a **project-owned C# simulator** using the same MQTTnet + Tahu-proto path the driver uses — validates encode/decode symmetrically and gives full control of rebirth/seq-gap/death scenarios (the §3.6 test matrix). Alternatives: Eclipse Tahu reference (Java/Python) edge-node script; Bevywise MQTT/IoT Simulator (purpose-built Sparkplug B). The simulator must publish NBIRTH/DBIRTH → periodic NDATA/DDATA, honour a rebirth NCMD, and be able to inject a seq gap / alias reuse / N-D-DEATH on demand. - **Plain-MQTT fixture:** a tiny publisher container (`mosquitto_pub` loop or a `paho-mqtt` script) emitting JSON on a handful of topics with `retain=true` so the last-value/read path **and** the `#`-observation browse are testable. - **Env-gated live suite:** `Driver.Mqtt.IntegrationTests`, category `LiveIntegration` (à la HistorianGateway), skips cleanly when **`MQTT_FIXTURE_ENDPOINT`** is unset → macOS-offline-safe. Covers: connect/TLS/auth; plain subscribe+retained-read; Sparkplug birth→data→alias-resolve→`OnDataChange`; rebirth recovery; death→STALE; the browser observation window (passive birth-driven tree accumulation — assert no NCMD is published by open/browse calls — plus the explicit `RequestRebirthAsync` enumeration, scoped to node vs. group). - **Live-verify (docker-dev):** author an Mqtt driver + tag via the `/uns` picker on `:9200`, deploy, confirm the node carries live broker values and the typed editor renders both modes (the usual live-`/run` discipline — Razor binding bugs pass unit tests). Add the fixture endpoint to `CLAUDE.md`'s endpoint list once landed (e.g. `10.100.0.35:1883`/`8883`, `MQTT_FIXTURE_ENDPOINT`). --- ## 10. Phasing + effort Order: **plain MQTT first, then Sparkplug B, then write.** Plain MQTT is the dependency-light slice that stands up the whole skeleton and **validates MQTTnet-v5-on-net10 before any Sparkplug/proto work** — front-loading the reusable plumbing and de-risking the library decision. | Phase | Scope | Rel. effort | |---|---|---| | **P1 — Plain MQTT** | 3 projects; MQTTnet-5 connect/TLS/auth + the **hand-rolled reconnect loop** (backoff + resubscribe-on-reconnect + session-state handling, §8 — v5 has no `ManagedMqttClient`) in `MqttConnection`; `ISubscribable` (topic subscribe → `OnDataChange`); `IReadable` last-value (retained seed); authored-tag `ITagDiscovery` (`Once`); `IHostConnectivityProbe`; `MqttDriverProbe`; factory + host wiring; `#`-observation browser (`MqttDriverBrowser`/`MqttBrowseSession`); typed editor + validator + AdminUI driver page; Mosquitto + JSON-publisher fixtures + env-gated suite. **Validate MQTTnet-5/net10 here.** | **M** | | **P2 — Sparkplug B ingest** | Vendored Tahu proto + `Google.Protobuf` codegen; `SparkplugCodec` decode; `spBv1.0/{group}/#` subscribe; N/DBIRTH → `BirthCache`/`AliasTable`; bind-by-name + seq-gap detect + **rebirth NCMD**; N/DDEATH → STALE; STATE/primary-host option; `ITagDiscovery` `UntilStable` + `IRediscoverable`; passive birth-driven browser tree + `AttributesAsync` + the operator **"Request rebirth"** picker action (§4.1); Sparkplug datatype map; C# edge-node simulator fixture + §3.6 test matrix. | **L** | | **P3 — Write-through (optional)** | `IWritable`: Sparkplug NCMD/DCMD + plain publish; optimistic-Good + self-correct on echo; `WriteIdempotent` respect (default off for non-idempotent Sparkplug commands); write-topic config. | **M** | **Top risks:** (1) **Sparkplug state-machine correctness** (aliases + rebirth + seq — §3.6; bind-by-name is the mitigation, test against a controllable simulator); (2) **library/dependency friction** — SparkplugNet's MQTTnet-4.x transitive pin ✕ this repo's fragile central pinning (transitive pinning off), mitigated by **hand-rolling the Tahu proto over one MQTTnet-5 client** (§2.1). Secondary: passive-browse coverage gaps (the explicit "Request rebirth" action + manual entry — the window itself never publishes), unbounded address space (authored-only runtime discovery), write-has-no-ack (accepted, optimistic-Good like Galaxy), broker security (TLS + real auth enforced).