diff --git a/code-reviews/Driver.Mqtt/findings.md b/code-reviews/Driver.Mqtt/findings.md new file mode 100644 index 00000000..c5dfd3de --- /dev/null +++ b/code-reviews/Driver.Mqtt/findings.md @@ -0,0 +1,340 @@ +# 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 | 2 | + +## 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.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 | Open | + +**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:** _(open)_ + +### 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 | Open | + +**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:** _(open)_ + +### 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.