# Code Review — Driver.Mqtt

| Field | Value |
|---|---|
| Module | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt` |
| Reviewer | Claude Code |
| Review date | 2026-08-10 |
| Commit reviewed | `b990adae` |
| Status | Reviewed |
| Open findings | 0 |

## Checklist coverage

A comprehensive review completes every category, recording "No issues found" where
a category produced nothing rather than leaving it blank.

| # | Category | Result |
|---|---|---|
| 1 | Correctness & logic bugs | Driver.Mqtt-001, Driver.Mqtt-003 |
| 2 | OtOpcUa conventions | No issues found |
| 3 | Concurrency & thread safety | Driver.Mqtt-002, Driver.Mqtt-006 |
| 4 | Error handling & resilience | Driver.Mqtt-004, Driver.Mqtt-007 |
| 5 | Security | No issues found |
| 6 | Performance & resource management | Driver.Mqtt-002 (leaked broker session per re-initialize) |
| 7 | Design-document adherence | Driver.Mqtt-001 (docs/drivers/Mqtt.md's "ingest state is reset at every session seam" was not true of the plain path) |
| 8 | Code organization & conventions | Driver.Mqtt-005 |
| 9 | Testing coverage | Driver.Mqtt-008 |
| 10 | Documentation & comments | Driver.Mqtt-009, Driver.Mqtt-010 |

#### Scope

The 11 hand-written source files under the module root and `Sparkplug/` (~8,100 LOC):
`MqttDriver`, `MqttConnection`, `MqttSubscriptionManager`, `LastValueCache`,
`MqttDriverFactoryExtensions`, `MqttDriverProbe`, and `Sparkplug/{AliasTable, BirthCache,
SequenceTracker, SparkplugIngestor, SparkplugPrimaryHost}`. Nothing under `bin/` or `obj/` was
reviewed (generated build output). `.Contracts` and `.Browser` are separate modules, read only
where this module's behaviour depends on them.

#### What is already right, and worth not regressing

Recorded because several of these are the reason a defect below is a defect in one narrow spot
only: the refused-CONNACK classification (MQTTnet 5 returns rather than throws), a deadline on
every wait including teardown, the RawPath-as-published-reference discipline in both ingest
paths, the alias-vs-name binding rule, the `Reinterpret` two's-complement call, "an invalid
payload never mutates state", the per-node rebirth debounce *and* its lifetime cap, and the
dispatcher-thread containment (nothing throws, blocks, or does I/O on MQTTnet's pump).

## Findings

### Driver.Mqtt-001

| Field | Value |
|---|---|
| Severity | High |
| Category | Correctness & logic bugs |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs:193` (`AttachTo`), `:328` (`SubscribeAsync`'s pending set) |
| Status | Resolved |

**Description:** In Plain mode the manager dedupes SUBSCRIBEs against `_subscribedTopics` — a set
that describes **the session it established them on** — but `AttachTo` never cleared it, and the
manager *outlives* a session rebuild. `MqttDriver.AdoptOptions` replaces the manager only when the
ingest identity changed, so both routes that rebuild the broker session re-attach the *same*
instance to a *new* connection: `ReinitializeAsync`'s rebuild branch (endpoint / credentials / TLS /
timeout change — `SessionIdentity` is a strict superset of `IngestIdentity`) and, more importantly,
the driver-host's `ShutdownAsync` + `InitializeAsync` retry, which the #527 comms-loss watchdog now
makes automatic and unattended.

After such a rebuild the host re-applies its desired subscriptions
(`DriverInstanceActor.InitializeSucceeded` → `ResubscribeDesired`), `SubscribeAsync` computed
`pending = defs.Where(d => !_subscribedTopics.ContainsKey(d.Topic))` — empty, because every topic was
still marked established from the *dead* session — and `EstablishAsync([])` returned immediately
having sent nothing. The result is exactly the failure this file's own remarks exist to prevent: a
driver reporting `Connected` / `Healthy` on a broker session carrying **zero subscriptions**,
receiving nothing, with no error, no exception and no Bad status. It self-corrected only if the
*new* session later dropped, at which point `OnReconnectedAsync` finally re-established the filters —
so a clean reconnect left every value this driver serves frozen indefinitely.

`SparkplugIngestor.AttachTo` already performed the equivalent reset (`ResetSessionState`) for its own
per-session state, with remarks spelling out this same "the ingestor outlives a session rebuild"
reasoning. The plain path was never given the same treatment.

**Recommendation:** Clear `_subscribedTopics` at the top of `AttachTo`, before delivery is wired,
mirroring `SparkplugIngestor.AttachTo`. Pin it with a test that rebinds the manager to a new
connection and asserts the next subscribe actually reaches the transport, plus a control proving the
within-one-session dedupe still holds — otherwise the "fix" is indistinguishable from deleting the
dedupe, which re-triggers every topic's retained seed on every subscribe.

**Resolution:** Fixed `2b4b3fcb`, 2026-08-10 — `AttachTo` forgets the previous session's established
filters. Pinned by `MqttSubscriptionManagerTests.AttachTo_ForgetsTheFiltersEstablishedOnThePreviousSession`
(confirmed to fail against the pre-fix code) with `SubscribeAsync_TwiceOnTheSameSession_StillIssuesOneSubscribe`
as the falsifiability control.

### Driver.Mqtt-002

| Field | Value |
|---|---|
| Severity | High |
| Category | Concurrency & thread safety |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs:1000` (`_connection = connection`), `:1297` (`_primaryHost = host`) |
| Status | Resolved |

**Description:** `InitializeCoreAsync` published `_connection` and `_primaryHost` — both
`IAsyncDisposable`, both owning a live broker socket and a running reconnect supervisor — **over
whatever the driver already held**, without disposing it. A second initialize on a live driver
therefore stranded a fully connected, self-supervising session that no later `ShutdownAsync` or
`DisposeAsync` could reach (both act only on the current field value), and whose own supervisor kept
it alive against the broker indefinitely. For a Sparkplug driver the orphan also retains its
group-wide `spBv1.0/{group}/#` subscription and a *second* primary-host claim on the same host id —
and two MQTT clients presenting one client id take turns evicting each other, forever.

CLAUDE.md records this as latent on the grounds that `DriverInstanceActor` passes `teardownFirst` on
the #527 retry path. That covers the retry path only: `AdoptConfigDuringInit` re-runs
`InitializeAsync` with **no** teardown when an `ApplyDelta` arrives while the driver is in
`Connecting`/`Reconnecting` (`DriverInstanceActor.cs:743`) — an ordinary redeploy racing a reconnect.
Depending on every caller to tear down first is also the wrong shape: the driver owns those fields
and is the only thing that can dispose them.

**Recommendation:** `await TeardownAsync()` at the top of `InitializeCoreAsync`, under the lifecycle
gate both callers already hold. It is idempotent (`Interlocked.Exchange` to null), so
`ReinitializeAsync`'s rebuild branch tearing down first costs a no-op and a first initialize sees
three nulls.

**Resolution:** Fixed `2b4b3fcb`, 2026-08-10 — `InitializeCoreAsync` tears down the previous
connection, host-probe loop and primary-host claim before building the new session. Pinned by
`MqttDriverSessionLifecycleTests.InitializeAsync_OnALiveDriver_TearsDownThePreviousSession`, which
asserts the **broker's** live socket count (a leak is by definition the case where the driver's own
view disagrees with the broker's) and fails against the pre-fix code.

### Driver.Mqtt-003

| Field | Value |
|---|---|
| Severity | Medium |
| Category | Correctness & logic bugs |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs:376,903`; `MqttConnection.cs:474,486,657,1039,1134,1368,1384`; `Sparkplug/SparkplugPrimaryHost.cs:267` |
| Status | Resolved |

**Description:** `MqttDriverOptions.ConnectTimeoutSeconds` carries `[Range(1, int.MaxValue)]`, but
nothing validates it on the runtime path — the blob is deserialized straight out of the deploy
artifact by `MqttJson.Options`. Every bounded wait in the module then feeds it to
`new CancellationTokenSource(TimeSpan.FromSeconds(...))`, `SemaphoreSlim.WaitAsync(TimeSpan)` or
`Task.WaitAsync(TimeSpan)` unchecked:

- `0` yields an already-cancelled deadline, so connect / subscribe / publish fail instantly against a
  perfectly healthy broker and the driver reports a timeout it invented;
- a negative value is outside the legal range of both APIs, so it throws
  `ArgumentOutOfRangeException` out of `ShutdownAsync` and `DisposeAsync` — the two paths whose whole
  job is to complete, and the ones a host calls while tearing a bad deployment down.

The module already clamps its other numeric knobs at their use sites (`MaxPayloadBytes` in the
manager's constructor, min/max backoff in `NextBackoff`, QoS in `MapQos`, and the probe's own
`ConnectTimeoutSeconds > 0 ? … : fallback`), so this knob is the outlier rather than the convention.

**Recommendation:** Normalize once, at the single seam every options instance the driver holds passes
through, rather than at each of the ten use sites — the value rides into `MqttConnection`,
`SparkplugPrimaryHost` and `MqttClientOptionsFactory` through `_options`.

**Resolution:** Fixed `2b4b3fcb`, 2026-08-10 — `MqttDriver.Normalize` replaces a non-positive
`connectTimeoutSeconds` (and a negative `keepAliveSeconds`; zero keep-alive is legal MQTT) with the
record's own defaults, applied in the constructor and in `ParseOptions`. Pinned by
`MqttDriverSessionLifecycleTests.ShutdownAndDispose_WithANegativeConnectTimeout_DoNotThrow`, with
`InitializeAsync_WithAnAuthoredConnectTimeout_KeepsIt` as the control that a legal value is not
quietly rewritten to the default.

### Driver.Mqtt-004

| Field | Value |
|---|---|
| Severity | Low |
| Category | Error handling & resilience |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs:375` (`ShutdownAsync`) |
| Status | Resolved |

**Description:** `ShutdownAsync` captured the last-message instant with `ReadHealth()` — the *stored*
snapshot, which is only ever written at initialize/shutdown — rather than `GetHealth()`, which
prefers the live `MqttConnection.LastMessageUtc`. The stored value is whatever was true when the
session was established (usually `null`), so a shutdown discarded the evidence that the driver had
been delivering right up to teardown and reported a driver that was receiving a second ago as one
that had never received anything. That timestamp is what the #527 comms-loss watchdog measures a
driver against, and shutdown-then-initialize is precisely its retry cycle.

**Recommendation:** Use `GetHealth().LastSuccessfulRead`; the connection is still live at that point
in the method.

**Resolution:** Fixed `2b4b3fcb`, 2026-08-10 — one-line change plus a comment naming why the two
health reads are not interchangeable here.

### Driver.Mqtt-005

| Field | Value |
|---|---|
| Severity | Low |
| Category | Code organization & conventions |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs:1-4` |
| Status | Resolved |

**Description:** Three `using` directives left behind when the TLS / CA-pin code moved out to the
`.Transport` module (Gitea #512) — `System.Net.Security`, `System.Security.Cryptography` and
`System.Security.Cryptography.X509Certificates` — no longer resolve anything in this file. They read
as though this type still does certificate work, which is exactly the layering claim #512 removed.
(`System.Buffers` looks equally unused and is not: `ReadOnlySequence<byte>.ToArray()` is
`BuffersExtensions`', not a member — removing it breaks the build.)

**Recommendation:** Drop the three, and leave a note on `System.Buffers` so the next reader does not
repeat the experiment.

**Resolution:** Fixed `2b4b3fcb`, 2026-08-10 — the three unused usings removed; `System.Buffers` kept
with a comment naming the extension method that needs it. Build verified.

### Driver.Mqtt-006

| Field | Value |
|---|---|
| Severity | Low |
| Category | Concurrency & thread safety |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs:278` |
| Status | Resolved |

**Description:** `DisposeAsync` calls `_gate.Dispose()`. `StopAsync` cancels the reconcile loop but
does not await it, so a `ReconcileAsync` already in flight can be sitting in
`ClaimAsync`/`ReleaseAsync`'s `_gate.WaitAsync(...)`, or on its way to the
`finally { _gate.Release(); }`, when the gate is disposed — surfacing an `ObjectDisposedException`
naming `SemaphoreSlim`, which tells an operator nothing about what happened. It is contained (both
`ReconcileAsync` and `ReconcileLoopAsync` catch broadly and log) and no claim can be created after
dispose (`ClaimAsync` re-checks `_disposed` under the gate), so the impact is a misleading warning
rather than a leak.

Worth recording because the sibling type reached the opposite conclusion **deliberately**:
`MqttConnection` documents that its `_lifecycleGate`, `_reconnectWake` and `_lifetimeCts` are *never*
disposed, precisely so a late callback racing teardown produces a benign no-op instead of an
`ObjectDisposedException` naming the wrong type. Two lifecycle types in one module disagreeing about
this is the kind of divergence that gets copied into the third.

**Recommendation:** Either stop disposing `_gate` (matching `MqttConnection`'s documented rule and its
stated rationale — the gate holds no timer and no surviving registration), or have `StopAsync` await
the reconcile loop before dispose returns. The former is the smaller change and the one already
argued for in this module.

**Resolution (2026-08-10):** Fixed `3f85ba98` — the first option. `DisposeAsync` no longer disposes
`_gate`, and both the field and the dispose site now carry the rule and the reason, so the module
states **one** answer for its lifecycle primitives rather than two. Nothing is leaked (the gate
holds no timer and no registration) and `ClaimAsync` still re-checks `_disposed` under it, so no
claim can be created after dispose either way. TDD: a test that disposes while a reconcile is in
flight failed RED with `log.Captured should not contain an element satisfying (name ==
"ObjectDisposedException") but does`. It is deterministic rather than timing-dependent — the gate is
taken *before* the connect starts, so a `MiniBroker` in blackhole mode (TCP accepted, CONNECT never
answered) having accepted the socket proves the claim is inside the gate when dispose runs. Suite
green (734).

### Driver.Mqtt-007

| Field | Value |
|---|---|
| Severity | Low |
| Category | Error handling & resilience |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs:406-424` (`OnReconnectedAsync`) |
| Status | Resolved |

**Description:** When the reconnect path finds live subscribed topics but resolves **zero** filters
for them, it logs a warning and returns normally. The type's own contract, stated two paragraphs
later, is that a reconnect which re-establishes nothing must *throw*, so `MqttConnection` tears the
fresh session down and retries rather than serving a connected-but-deaf driver — and this branch
reaches that same outcome by a different route (no subscription is re-established either way). The
asymmetry is deliberate as written ("Silence here is how the wildcard defect hid"), but the remedy it
chose was a louder log rather than the teardown the failure warrants.

In practice the branch is close to unreachable — `Register` prunes `_subscribedTopics` of exactly the
topics missing from `ByFilter`, and with Driver.Mqtt-001 fixed a rebind clears the set outright —
which is an argument for leaving the code alone, not for leaving the contract ambiguous.

**Recommendation:** Throw here as the zero-granted case does, or state explicitly in the remarks why
*this* deaf-reconnect shape is exempt from the type's own rule. Do not leave the two answers side by
side unexplained.

**Resolution (2026-08-10):** Fixed `2298485b` — it throws, and the remarks now state the single rule.
Exempting it was considered and rejected: the reason the partial-rejection case is exempt is that
tearing the session down would punish the tags still flowing, and here there are none — every live
topic is in the set that resolved to nothing — so the teardown costs nothing and the retry is not
futile, since it runs against whatever `Register` has published by then. The `subscribed.Count == 0`
branch stays a debug no-op: a reconnect with nothing subscribed yet re-establishes nothing because
there is nothing to establish. TDD: `Should.ThrowAsync<InvalidOperationException>` failed RED
(`should throw System.InvalidOperationException but did not`). The test reaches the state the only
way production can — `Register` prunes `_subscribedTopics` of the topics its new table dropped, but a
SUBSCRIBE already in flight records its own topic *after* that prune, so a deploy landing inside the
round-trip leaves a live topic no authored tag resolves to — driven through a `DuringSubscribe` hook
on the existing fake transport. Paired with the control that a reconnect with no live subscriptions
stays a no-op, which a blanket `filters.Count == 0` throw would have broken. Suite green (734).

### Driver.Mqtt-008

| Field | Value |
|---|---|
| Severity | Low |
| Category | Testing coverage |
| Location | `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/` (708 tests at `b990adae`) |
| Status | Resolved |

**Description:** Coverage is genuinely strong (golden Sparkplug vectors, the §3.6 message matrix,
CONNACK classification, backoff, retained-seed handling, write encoding, the accept-gate). The gap
that mattered is that **`MqttSubscriptionManager.AttachTo` had no test at all** — grepping `AttachTo`
across the suite matched only `SparkplugIngestorTests`. That is why Driver.Mqtt-001 survived: the
Sparkplug half of the same seam is pinned by `AttachTo_DropsBirthsFromThePreviousSession`, and the
plain half was not. Nothing covered a *driver-level* second initialize either (Driver.Mqtt-002),
because every driver-lifecycle test initializes exactly once.

The shape of the gap is worth naming: both defects live in "what happens to state that outlives the
session it describes", and both are invisible to a suite whose fixtures build one session per test.

**Recommendation:** Cover every session-rebind seam on *both* ingest paths, and assert
re-initialization against an observable the driver does not own (the fake broker's live socket
count), since a leak is by definition the case where the driver's own view is wrong.

**Resolution:** Fixed `2b4b3fcb`, 2026-08-10 — added
`MqttSubscriptionManagerTests.AttachTo_ForgetsTheFiltersEstablishedOnThePreviousSession` and
`…SubscribeAsync_TwiceOnTheSameSession_StillIssuesOneSubscribe`, plus the new
`MqttDriverSessionLifecycleTests` (3 tests). All three regression tests were confirmed to fail
against the un-fixed production code, and both controls to pass in either state. Suite 708 → 713
green.

### Driver.Mqtt-009

| Field | Value |
|---|---|
| Severity | Low |
| Category | Documentation & comments |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs:241-248` (type remarks) |
| Status | Resolved |

**Description:** The remarks on the publish leg stated "This driver has no `IWritable` leg in v1, so
a Sparkplug rebirth NCMD is the **only** outbound application message it ever sends." Gitea #508
added the write leg — `MqttDriver.WriteAsync` calls `connection.PublishAsync` for every OPC UA write
— and #511 added a third caller in `SparkplugPrimaryHost`'s retained STATE. A comment that
under-states what a method is used for is how a later "this is only ever called from X, so it can
assume Y" change gets made.

**Recommendation:** Enumerate the three production callers instead, and date the "no `IWritable` leg"
claim as historical.

**Resolution:** Fixed `2b4b3fcb`, 2026-08-10 — remarks now name all three callers and mark the
retired claim against #508.

### Driver.Mqtt-010

| Field | Value |
|---|---|
| Severity | Low |
| Category | Documentation & comments |
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs:294` (`Reset`) |
| Status | Resolved |

**Description:** `Reset()`'s remarks call it "the reconnect seam (§3.6 invariant #5 — late join)",
but the reconnect path does not call it: `SparkplugIngestor.ResetSessionState` clears the whole
`_trackers` map instead. Its only caller in the tree is `SequenceTrackerTests`. The reasoning in the
remarks is correct and load-bearing — it is the attribution that is wrong, and it sends a reader
looking for the reconnect reset to the wrong method.

**Recommendation:** Keep the method and the rationale; state that the ingestor drops the map rather
than resetting instances, and why (dropping also releases the entry for a node the plant stopped
publishing).

**Resolution:** Fixed `2b4b3fcb`, 2026-08-10 — remarks corrected; no code change.