lmxopcua/docs/drivers/Mqtt.md

MQTT Driver

In-process MQTT broker-subscribe driver. It holds one MQTTnet 5 client against one broker, subscribes to the topics its authored tags name, and forwards each PUBLISH straight to ISubscribable.OnDataChange. Ingest is push, not poll, and one-way — there is no IWritable.

Two ingest shapes, selected by the driver's Mode: Plain binds a tag to a concrete MQTT topic; SparkplugB binds it to a group / edge node / device? / metric tuple and decodes Eclipse-Tahu protobuf under one spBv1.0/{GroupId}/# subscription. Both run over the same single client.

See Mqtt-Test-Fixture.md for the harness and ../plans/2026-07-15-mqtt-sparkplug-driver-design.md for the design.

Status. P1 (plain MQTT) and P2 (Sparkplug B ingest) are both shipped and live-gated against the Mosquitto + Sparkplug-simulator fixture. Write-through (IWritable — NCMD/DCMD/plain publish) is deferred; every MQTT node materializes read-only. See Known gaps for what is wired but not yet end-to-end observable.

Project Layout

Project	Holds
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.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 (deferred) is a leaf Driver.Mqtt.Transport project.

Capability Surface

public sealed class MqttDriver
    : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable, IAsyncDisposable, IDisposable

Capability	Entry point	Notes
IDriver	InitializeAsync / ReinitializeAsync / ShutdownAsync / GetHealth	Order register → attach → connect is contractual. ReinitializeAsync applies a tag-only delta in place, and rebuilds the session when the broker identity changed
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
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

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."

Driver-type string: Mqtt (DriverTypeNames.Mqtt).

Configuration

Bound through MqttJson.Options — case-insensitive, enums by name, unknown members skipped.

The whole connection is authored on the DRIVER (DriverConfig), via MqttDriverForm in the /raw driver-config modal. Unlike Modbus / S7 / OPC UA Client, MQTT does not use the v3 endpoint→DeviceConfig split: an MQTT driver instance holds exactly one broker session, so MqttDeviceForm is informational (the same shape as GalaxyDeviceForm) and devices under an MQTT driver are pure organisational grouping in the RawPath.

Why not the device. DriverDeviceConfigMerger merges a device's keys up to the top level only when the driver has exactly one device. A device-authored broker connection therefore disappears from the merged blob the moment a second device is added — a silent outage with nothing to see at deploy time. Driver-level authoring is correct for 0, 1 or N devices. A pre-form deployment that put the connection on a sole device still works (merge-up wins over the driver's keys) and MqttDeviceForm flags those keys so the mismatch is visible; move them to the driver.

DriverDeviceConfigMerger still merges driver + device and injects rawTags.

// DriverConfig — authored by MqttDriverForm. PascalCase keys, enums as names.
{
  "Host": "10.100.0.35",
  "Port": 8883,
  "UseTls": true,
  "AllowUntrustedServerCertificate": false,
  "CaCertificatePath": null,      // pins the chain; null = OS trust store
  "Username": "otopcua",
  "Password": "",                 // never commit
  "ProtocolVersion": "V500",
  "CleanSession": true,
  "KeepAliveSeconds": 30,
  "ConnectTimeoutSeconds": 15,
  "ReconnectMinBackoffSeconds": 1,
  "ReconnectMaxBackoffSeconds": 30,
  "MaxPayloadBytes": 1048576,
  "Mode": "Plain",
  "Plain": { "TopicPrefix": "", "DefaultQos": 1 }
}

// DeviceConfig — empty; MQTT has no per-device endpoint.
{}

Sparkplug sub-object (Mode: "SparkplugB")

{
  "Mode": "SparkplugB",
  "Sparkplug": {
    "GroupId": "OtOpcUaSim",            // REQUIRED — the driver's whole subscription scope
    "HostId": "",                       // Host Application identity; receive-only (see below)
    "ActAsPrimaryHost": false,          // NOT IMPLEMENTED — logs a warning when true
    "RequestRebirthOnGap": true,        // a seq gap asks the edge node to re-announce (NCMD)
    "BirthObservationWindowSeconds": 15 // how long discovery collects births before calling it stable
  }
}

Key	Default	Notes
GroupId	""	Required. The driver subscribes exactly spBv1.0/{GroupId}/#. Blank ⇒ it connects, reports Healthy, and ingests nothing. It is a literal topic segment, so /, +, # are refused by the form
HostId	""	Sparkplug Host Application identity. Receive-only — the driver observes STATE and never publishes one
ActAsPrimaryHost	false	Not implemented. Setting it logs a startup warning rather than being silently inert — being a primary host is a three-part contract (retained ONLINE, a matching OFFLINE registered as the MQTT Will at connect time, an explicit OFFLINE on clean shutdown), and shipping the 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
RequestRebirthOnGap	true	On a detected sequence-number gap, publish a rebirth-request NCMD instead of running on stale metric state
BirthObservationWindowSeconds	15	Discovery's birth-collection window. Must be ≥ 1

MqttDriverForm authors all five in Sparkplug mode. The sub-object is merged, never replaced (a key a newer driver adds inside it survives an older AdminUI), and in Plain mode it is not touched at all — so flipping a driver to Plain to look at it and back does not lose the group id. The form never writes RawTags; the deploy artifact owns it.

This was a live-gate finding. Through the P2 implementation MqttDriverForm still carried its P1 placeholder — switching Mode to SparkplugB rendered a "not available yet" notice and no group-id field, so once Sparkplug ingest shipped there was no way to author a Sparkplug driver from the AdminUI at all. Pinned by MqttDriverFormModelTests' Sparkplug cases.

Every key has a default, so nothing is strictly required by the binder. Notable ones: port 8883, useTls true, protocolVersion V500, cleanSession true, keepAliveSeconds 30, connectTimeoutSeconds 15, reconnectMinBackoffSeconds 1 / reconnectMaxBackoffSeconds 30, maxPayloadBytes 1 MiB, allowUntrustedServerCertificate false.

⚠️ Leave clientId unset on a redundant pair. Both nodes of a pair run the driver. MQTT requires client ids to be unique per broker, so a fixed clientId makes the two nodes evict each other forever — the broker logs Client <id> already connected, closing old connection and both nodes reconnect every few seconds while still reporting Healthy. Unset (the default) lets MQTTnet generate a unique id per connection, which is correct. This was observed live during the P1 gate. (The probe and browser already self-uniquify with -probe- / -browse- suffixes.) MqttDriverForm defaults the field blank and raises this warning inline the moment a value is typed.

The form validates every knob against the driver's own [Range] bounds (port 1–65535, keep-alive / connect-timeout / backoffs ≥ 1 s, max-backoff ≥ min-backoff, maxPayloadBytes ≥ 1, QoS 0–2) and additionally clamps on serialize, so an operator who ignores the inline error still cannot persist a connectTimeoutSeconds: 0 driver-brick.

Tag Configuration

A tag binds in one of two shapes, and which one is decided by which keys are present, not by a mode key on the tag. MqttTagDefinitionFactory reads the Sparkplug tuple when any Sparkplug descriptor key is set, and the topic otherwise; the AdminUI's "Tag shape" selector infers the same way on reopen. The key names are single-sourced in MqttTagConfigKeys so the browse-commit mapper and the factory cannot drift.

Plain shape

Key	Type	Default	Notes
topic	string	—	Required. Absent/blank rejects the tag ⇒ BadNodeIdUnknown
payloadFormat	Json | Raw | Scalar	Json	Strict — a typo'd value rejects the tag
jsonPath	string	"$"	Absent or empty ⇒ document root. Json only
dataType	DriverDataType	String	Strict. Float64, not Double — see below
qos	int 0–2	driver's defaultQos	Strict — a malformed value rejects rather than silently weakening delivery
retainSeed	bool	true	Seed this tag from the broker's retained message on subscribe

There is no mode key — the AdminUI's "Tag shape" selector is UI-only, inferred from the blob and never serialized.

dataType uses DriverDataType, which has Float64 and has no Double. Authoring "Double" is a strict-enum rejection, not a fallback.

Payload handling: Json selects via JSONPath then coerces (miss ⇒ BadDecodingError, coerce fail ⇒ BadTypeMismatch); Raw returns the payload's UTF-8 text verbatim and ignores dataType; Scalar decodes then coerces. The JSONPath subset is $, $.a.b, $['a'], $.a[0] — no filters, slices, wildcards or recursive descent.

Wildcards

The runtime accepts a +/# topic (it is ambiguous, not unparseable) and the deploy-time Inspect pass warns. The AdminUI editor rejects it outright — the one place the editor is deliberately stricter than the driver, because one Tag holds one value and a wildcard would feed it from many source topics. Wildcard tags are indexed by filter, not by concrete topic, so they survive a reconnect; a message fans out to every matching tag, and unauthored topics are dropped.

Sparkplug shape

Key	Type	Default	Notes
groupId	string	—	Required. Must match the driver's Sparkplug.GroupId to ever receive a message
edgeNodeId	string	—	Required.
deviceId	string	absent	Omit for a metric published by the edge node itself (NBIRTH/NDATA). Absent is meaningfully different from ""
metricName	string	—	Required. The tag's stable binding key across rebirths. May contain / — Node Control/Rebirth and Properties/Serial Number are canonical Sparkplug names, and the whole name is one value, never split
dataType	DriverDataType	from the birth	Optional override. Absent ⇒ take whatever type the birth certificate declared. The AdminUI's "(from birth certificate)" option removes the key entirely
isHistorized / historianTagname	—	—	Server-side historization, unchanged from Plain

payloadFormat, jsonPath, qos and retainSeed are Plain-only and meaningless here: a Sparkplug body is protobuf, and the subscription is one driver-wide spBv1.0/{GroupId}/#, not a per-tag filter.

groupId / edgeNodeId / deviceId are literal MQTT topic segments, so the AdminUI editor refuses /, + and # in them — a decoded incoming id can never contain one, so such a tag could never bind. metricName is deliberately exempt: it is not a topic segment.

// A device metric.
{"groupId":"OtOpcUaSim","edgeNodeId":"EdgeA","deviceId":"Filler1","metricName":"FillCount"}

// A node-level metric — note deviceId is ABSENT, not blank.
{"groupId":"OtOpcUaSim","edgeNodeId":"EdgeA","metricName":"Temperature"}

Sparkplug B Ingest

One subscription (spBv1.0/{GroupId}/#) feeds a per-edge-node state machine (SparkplugIngestor + SparkplugCodec / AliasTable / BirthCache / SequenceTracker):

• Births are the only self-describing message. NBIRTH/DBIRTH name every metric and assign its alias; NDATA/DDATA carry an alias and no name at all. So the alias table is rebuilt per birth and tags bind by name, never by alias — aliases are per-birth and may be reused for a different metric after a rebirth.
• NBIRTH is node-scoped, DBIRTH is device-scoped, and routing a DBIRTH through the node path wipes bdSeq. They are handled separately on purpose.
• Sequence tracking: seq is a wrapping 0–255 counter. A detected gap raises a rebirth request when RequestRebirthOnGap is on, rather than silently continuing on stale metric state.
• Death → STALE. An NDEATH whose bdSeq matches the live session (the bdSeq tie is what stops a stale death from killing a fresh session) marks every metric under that edge node Bad / BadNoCommunication. A DDEATH does the same for one device. The next birth restores them.
• Unsupported datatypes are skipped with a warning, never guessed: DataSet, Template, PropertySet, PropertySetList, Unknown. Int8→Int16 and UInt8→UInt16 widen; *Array variants map to the element type with the array bit set.
• Ingest state is reset at every session/authoring seam — a reconnect or a ReinitializeAsync must not carry a previous session's alias table into a new one.

Rediscovery (Sparkplug)

OnRediscoveryNeeded fires only when a birth is for an authored scope and its metric-name set differs from the last one seen for that scope (order-insensitive, ordinal). Both conditions are anti-storm, not optimisations: rediscovery triggers an address-space rebuild, and an edge node re-births freely — on its own schedule, on every reconnect, and once per rebirth NCMD.

⚠️ Nothing consumes the signal today. No runtime component subscribes to IRediscoverable.OnRediscoveryNeeded, and DriverHostActor.HandleDiscoveredNodes hard-returns — discovered-node injection is dormant in v3. The served address space comes from the deploy artifact, so a DBIRTH introducing a new metric does not change the OPC UA tree; you must redeploy. The driver's decision is observable only as an Information log line (… requesting rediscovery.). This is a v3 platform gap, not an MQTT one.

Connection + Failure Semantics

MqttConnectionState: Disconnected → Connected → Reconnecting → Faulted → Disposed. A broker that is merely down stays Reconnecting indefinitely — unreachable is never a fault. Backoff is exponential between the two reconnectBackoff bounds.

MQTTnet 5 does not throw on a rejected CONNACK — it returns the code and leaves the client disconnected. Left unhandled, a wrong password produced a Healthy driver that never received a value. MqttConnection therefore raises MqttConnectRejectedException:

• Unrecoverable ⇒ Faulted, supervisor stops (retrying cannot help): MalformedPacket, ProtocolError, UnsupportedProtocolVersion, ClientIdentifierNotValid, BadUserNameOrPassword, NotAuthorized, Banned, BadAuthenticationMethod, TopicNameInvalid, PacketTooLarge, PayloadFormatInvalid, RetainNotSupported, QoSNotSupported, ServerMoved.
• Everything else is transient and keeps retrying — including UnspecifiedError and any future code. (ServerMoved is permanent, UseAnotherServer is temporary — per the MQTT 5 spec.)

A reconnect that completes without re-subscribing would be a healthy-looking connection that receives nothing forever, so OnReconnectedAsync throws when zero filters are granted, tearing the session down rather than swallowing it. A SUBACK failure degrades just the affected RawPaths to BadCommunicationError.

Browse

MqttDriverBrowser opens a read-only observation session: it subscribes at QoS 0 to # (or {topicPrefix}/#) under a -browse--suffixed client id and builds a /-segment tree from whatever publishes. Nothing is ever published. It is inherently incomplete — a topic that stays silent during the window is invisible — so manual entry remains the escape hatch.

Bounds: 50 000 nodes (then ⚠ Observation truncated), 1024-char topics / 256-char segments (then ⚠ Some topics were too long), 512-byte payload inspection. Sessions are reaped after 2 minutes idle.

Sparkplug B browse serves a different tree from the same passive window: observed NBIRTH/DBIRTH certificates are decoded into Group → EdgeNode → [Device] → Metric (a birth is the only Sparkplug message that names its metrics). Same node cap, same bounds, same "publishes nothing" guarantee.

The bespoke browser wins over the universal DiscoveryDriverBrowser, which would decline MQTT anyway (SupportsOnlineDiscovery is false).

Browse-commit writes the address the factory reads

A committed leaf's address is a descriptor, not a single reference string, so RawBrowseCommitMapper reads it from the AttributeInfo.AddressFields the session states — Plain emits topic; Sparkplug emits groupId / edgeNodeId / deviceId? / metricName, the names MqttTagConfigKeys single-sources for both the mapper and MqttTagDefinitionFactory.

It is never parsed out of the browse node id. A Sparkplug node id is {group}/{node}[/{device}]::{metric} and a metric name legitimately contains / (Node Control/Rebirth), so splitting the id cannot recover the tuple. Which keys the session emits is also how the mapper learns the shape — the driver type reaching it is just Mqtt, and the mode lives on the driver config. A leaf whose attribute lookup returned nothing carries no address and is refused at commit, in words, rather than committed as a tag that deploys clean and then reports BadNodeIdUnknown forever.

Request rebirth (Sparkplug only)

Births are never retained, so a healthy but quiet plant shows an empty picker tree until one lands. The Request rebirth affordance in the browse modal is the way to fill it on demand: it publishes a Sparkplug rebirth-request NCMD (Node Control/Rebirth) to the selected scope — the one and only browse action that publishes anything.

• Scope is the last node clicked in the tree. A device or metric resolves up to its owning edge node (Sparkplug has no device-scoped rebirth); a group fans out to every observed edge node beneath it and is refused whole past 32, publishing nothing.
• Two clicks, never one — Request rebirth… arms a confirm panel naming the resolved scope and its consequence; changing the selection disarms it.
• Gated on DriverOperator, enforced server-side in BrowserSessionService.RequestRebirthAsync (fail-closed) before the session is touched; the UI gate is defence in depth. A refusal — policy, over-wide group, unresolvable scope — is shown, not swallowed.
• Offered only for a Sparkplug session (IRebirthCapableBrowseSession.RebirthAvailable); a Plain MQTT window publishes nothing, ever.

Refresh

An observation window accumulates, but the tree renders once, when it is created — and OpenAsync returns as soon as the SUBSCRIBE is granted, without waiting for the birth-observation window. So the first render is normally empty on Plain (a topic that has not published yet is invisible) and almost always empty on Sparkplug (births are never retained). Refresh re-reads the root against the same session: nothing reconnects, nothing already observed is lost, and the tag selection is kept. Only the armed rebirth scope is cleared, because the node it named may no longer be in the rebuilt tree.

Also fixed by this gate, and not MQTT-specific: every node label in the shared DriverBrowseTree was an <a href="#">. blazor.web.js's enhanced-navigation click interceptor resolves a bare # against <base href="/">, so clicking a node label navigated the whole AdminUI to / and destroyed the hosting modal (@onclick:preventDefault suppresses the browser's default action, not Blazor's interceptor). The labels are now <button type="button">. This affected every driver's picker since 2026-05-28; it surfaced here because choosing a rebirth scope is the first flow that requires clicking a label rather than the ▶ toggle.

⚠️ Known gap — a completely empty tree cannot arm a rebirth. The rebirth scope is taken from the last node clicked in the tree, so on a plant that has not birthed since the window opened there is nothing to click and Request rebirth… stays disabled — the exact case the affordance exists for. MqttBrowseSession.ResolveRebirthTargets already accepts a bare {group}/{edgeNode} for a node the window has never seen (it calls that "the prime rebirth target"); what is missing is a UI path to enter one, or to default the scope to the driver's own configured GroupId. Until then: use Refresh after any birth, or author the tags by Manual entry, which is always available.

Test Connect

MqttDriverProbe performs one CONNECT (no SUBSCRIBE, no publish) under a -probe- client id and disconnects. Success message: "MQTT CONNECT OK". A rejected CONNACK is described by the same DescribeConnackRejection the running driver uses, so the button and the driver report a rejection in identical words. See TestConnectProbes.md.

Testing

• Unit — tests/Drivers/…Driver.Mqtt.Tests (581 tests): tag-config mapping (both shapes), JSONPath subset, strict-enum rejection, subscription indexing (incl. the wildcard-by-filter case), reconnect/rebuild branches, CONNACK classification, and the Sparkplug half — golden-payload decode, topic parse/format, datatype map, alias/birth-cache rebuild, sequence + bdSeq-death ties, the rediscovery change gate, and the browse session's "publishes nothing" assertion.
• Live (env-gated) — tests/Drivers/…Driver.Mqtt.IntegrationTests (15 tests) against the Mosquitto broker + the C# Sparkplug edge-node simulator; skips cleanly when MQTT_FIXTURE_ENDPOINT is unset. See Mqtt-Test-Fixture.md and infra/README.md §3.
• AdminUI — MqttDriverFormModelTests / MqttTagConfigModelTests cover both editors' round-trip, mode inference and validation. AdminUI has no bUnit, so the razor shells themselves are only ever proven by a live /run gate.

Operational Notes

• Read-only. No writes reach the broker; every node is ViewOnly. Sparkplug NCMD is used for rebirth requests only — never to command a device.
• No alarms, no driver-side history. Historization is a server-side concern (isHistorized).
• Both pair nodes subscribe independently — MQTT fan-out is per-connection, and the Primary gate applies downstream, not to ingest. This is also why clientId must stay unset (above).
• Unauthored traffic is ignored. A busy broker is normal; the driver never auto-provisions a tag.

Known gaps

Gap	Effect	Notes
Write-through (IWritable)	Every MQTT node is read-only	Deferred to P3 — NCMD/DCMD + plain publish, optimistic-Good + self-correct on echo
Rediscovery is inert	A DBIRTH introducing a metric does not change the OPC UA tree; redeploy to pick it up	v3 platform gap — nothing subscribes to OnRediscoveryNeeded and DriverHostActor.HandleDiscoveredNodes hard-returns. Driver-side decision is log-observable
Rebirth needs a non-empty tree	Request rebirth… cannot be armed on a plant that has not birthed since the window opened	See Refresh. Session side already supports a bare {group}/{edgeNode} scope
_canRebirth is captured at browse-open	A session reaped for idleness (2 min) still draws the button, and using it errors	The error is shown, not swallowed; reopen the browser
Primary-host STATE publishing	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	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	Skipped with a warning	v1-unsupported; never coerced into a guessed type
The tag editor writes payloadFormat into a Sparkplug blob	Cosmetic only	payloadFormat is Plain-only; the factory routes on the Sparkplug keys and ignores it. Noise in the blob, not a mis-binding