Joseph Doherty
8d9155682d
Two gaps the Task 23/24 review found, both blocking Task 26's live gate.
Gap 1 — browse-commit produced a silently dead MQTT tag. RawBrowseCommitMapper
had no `Mqtt` case, so a committed leaf fell through to the generic
`{"address": …}` key. Neither MqttTagDefinitionFactory entry point reads
`address`: the tag deployed clean and reported BadNodeIdUnknown forever, with
no signal at commit time. Predates Task 23 (Plain was affected too), but Task 23
built the Sparkplug metric tree precisely so an operator could browse and commit
a binding.
The address is a DESCRIPTOR, not a reference string, and it cannot be recovered
from the browse node id: `{group}/{node}[/{device}]::{metric}` where a metric
name legitimately contains `/` (`Node Control/Rebirth`) — the ambiguity
MetricSeparator's remarks already name. So the session STATES it, via a new
`AttributeInfo.AddressFields` seam, and the mapper reads it. Which keys are
emitted is also how the mapper learns Plain vs Sparkplug — the driver type
reaching it is just `Mqtt`, and the mode lives on the driver config. Key names
are single-sourced in the new `MqttTagConfigKeys` (producer, mapper, factory),
with the literals pinned by a test so a symmetric rename cannot silently unbind
already-persisted blobs. A leaf with no stated address is refused at commit in
words rather than committed dead.
Gap 2 — RequestRebirthAsync had no UI. Task 23 shipped it backend-only; Task 26
step 1 assumes the button, and Task 25's runbook notes the picker tree stays
empty until a birth lands, so it is the only way to fill it on demand. Added to
the /raw browse modal: scope = the clicked tree node (device/metric resolve up to
their edge node, group fans out and is refused whole past 32), an explicit
two-click confirm naming the resolved scope and its consequence, the outcome or
the refusal shown rather than swallowed. Offered only for a Sparkplug session
(new `IRebirthCapableBrowseSession.RebirthAvailable` — one session class serves
both modes, so a type test alone would draw a button that can only throw) and
only to a DriverOperator; the server-side gate remains the boundary.
Tests: MQTT 545 → 581, AdminUI 781 → 800. The round-trip tests feed the emitted
blob to the REAL factory and were falsified by mutating the emitted key name.
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
239 lines
14 KiB
Markdown
239 lines
14 KiB
Markdown
# 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`.
|
||
|
||
Sparkplug B is **P2** and not implemented: `mode: SparkplugB` is a placeholder everywhere it appears.
|
||
See [Mqtt-Test-Fixture.md](Mqtt-Test-Fixture.md) for the harness and
|
||
[`../plans/2026-07-15-mqtt-sparkplug-driver-design.md`](../plans/2026-07-15-mqtt-sparkplug-driver-design.md)
|
||
for the design.
|
||
|
||
## 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
|
||
|
||
```csharp
|
||
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 => Once` — *a broker's live topic set is not a tag catalogue* |
|
||
| `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` | **Seam only** — nothing raises it in Plain mode. Sparkplug DBIRTH will |
|
||
|
||
**Not implemented:** `IWritable`, `IPerCallHostResolver`, `IAlarmSource`, `IHistoryProvider`.
|
||
Materialized nodes are `SecurityClassification.ViewOnly`, `IsAlarm: false`, `IsHistorized: false` —
|
||
*"MQTT ingest is one-way in P1 … 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`.
|
||
|
||
```jsonc
|
||
// 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.
|
||
{}
|
||
```
|
||
|
||
The form never writes `RawTags` (the deploy artifact owns it) and never touches `Sparkplug` (P2) — an
|
||
existing `Sparkplug` object, and any key a newer driver adds, survive a load→save untouched.
|
||
|
||
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
|
||
|
||
| 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.
|
||
|
||
## 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.
|
||
|
||
## 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](TestConnectProbes.md).
|
||
|
||
## Testing
|
||
|
||
- **Unit** — `tests/Drivers/…Driver.Mqtt.Tests` (266 tests): tag-config mapping, JSONPath subset,
|
||
strict-enum rejection, subscription indexing (incl. the wildcard-by-filter case), reconnect/rebuild
|
||
branches, CONNACK classification.
|
||
- **Live (env-gated)** — `tests/Drivers/…Driver.Mqtt.IntegrationTests` (7 tests) against the Mosquitto
|
||
fixture; skips cleanly when `MQTT_FIXTURE_ENDPOINT` is unset. See
|
||
[Mqtt-Test-Fixture.md](Mqtt-Test-Fixture.md) and `infra/README.md` §3.
|
||
|
||
## Operational Notes
|
||
|
||
- **Read-only.** No writes reach the broker; every node is ViewOnly.
|
||
- **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.
|
||
- **Unauthored traffic is ignored.** A busy broker is normal; the driver never auto-provisions a tag.
|