# 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.
Author the connection on the **device** (`DeviceConfig`) and the mode on the **driver**
(`DriverConfig`); `DriverDeviceConfigMerger` merges them, and injects `rawTags`.

```jsonc
// DriverConfig
{ "Mode": "Plain", "Plain": { "TopicPrefix": "", "DefaultQos": 1 } }

// DeviceConfig
{
  "Host": "10.100.0.35",
  "Port": 8883,
  "UseTls": true,
  "Username": "otopcua",
  "Password": "",                 // supply via ${secret:} / env — never commit
  "CaCertificatePath": null       // pins the chain; null = OS trust store
}
```

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

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