92ba120964
The P1 live gate had to insert MQTT driver config directly via SQL: DriverConfigModal and DeviceModal both fell through to "No typed config form for driver type Mqtt" with no raw-JSON fallback, so broker host/port/TLS/credentials were unauthorable from the AdminUI. Task 12 built the *tag* editor; these are the driver + device surfaces. MqttDriverForm authors the whole MqttDriverOptions connection surface (host/port/clientId, TLS + CA pin, credentials, protocol version/clean-session/keep-alive, connect timeout, reconnect backoffs, mode, maxPayloadBytes, and the Plain sub-object). The Sparkplug sub-object stays P2 — a marked placeholder mirroring MqttTagConfigEditor's stub, with any existing sparkplug keys preserved untouched. The connection is authored on the DRIVER, not the device — unlike Modbus/S7/OpcUaClient and contrary to what docs/drivers/Mqtt.md said. DriverDeviceConfigMerger merges a device's keys up only when the driver has exactly ONE device, so a device-authored broker connection vanishes silently the moment a second device is added. MqttDeviceForm is therefore informational (the GalaxyDeviceForm shape) and round-trips DeviceConfig verbatim; it flags legacy connection keys left on a device by a pre-form deployment, because those still win via merge-up. Serialization goes through the ONE shared MqttJson.Options instance (Task 9's decision), never a fifth per-form copy — pinned by an assertion that an ordinal-only reader CANNOT bind the emitted blob, and by extending the fleet-wide DriverPageJsonConverterTests guard to resolve a form's serializer from an explicit external-instance registry when it has no _jsonOpts field. Dropping the converter from MqttJson.Options reddens 3 tests and leaves the symmetric round-trip green — the Task 9 lesson, reproduced. RawTags is stripped from the emitted blob (the deploy artifact owns it) and unknown top-level keys survive a load->save. Validation mirrors the driver's own [Range] bounds inline, and ToOptions() additionally clamps, so an ignored error still cannot persist a connectTimeoutSeconds:0 driver-brick. A non-blank clientId raises the P1 live-gate warning inline (fixed ids make redundant pair nodes evict each other while both report Healthy). AdminUI 761/761 green (was 730), TWAE-clean; Driver.Mqtt 266/266 green. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
203 lines
12 KiB
Markdown
203 lines
12 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.
|
||
`SparkplugB` browse throws `NotSupportedException`.
|
||
|
||
The bespoke browser wins over the universal `DiscoveryDriverBrowser`, which would decline MQTT anyway
|
||
(`SupportsOnlineDiscovery` is `false`).
|
||
|
||
## 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.
|