diff --git a/code-reviews/Driver.Mqtt.Transport/findings.md b/code-reviews/Driver.Mqtt.Transport/findings.md new file mode 100644 index 00000000..8aa803d5 --- /dev/null +++ b/code-reviews/Driver.Mqtt.Transport/findings.md @@ -0,0 +1,132 @@ +# Code Review — Driver.Mqtt.Transport + +| Field | Value | +|---|---| +| Module | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport` | +| Reviewer | Claude Code | +| Review date | 2026-08-10 | +| Commit reviewed | `b990adae` | +| Status | Reviewed | +| Open findings | 1 | + +## 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.Transport-002 | +| 2 | OtOpcUa conventions | No issues found | +| 3 | Concurrency & thread safety | No issues found | +| 4 | Error handling & resilience | No issues found | +| 5 | Security | Driver.Mqtt.Transport-001 | +| 6 | Performance & resource management | No issues found | +| 7 | Design-document adherence | No issues found | +| 8 | Code organization & conventions | No issues found | +| 9 | Testing coverage | No issues found | +| 10 | Documentation & comments | No issues found | + +#### Scope + +One hand-written file, `MqttClientOptionsFactory.cs` (275 LOC), plus the csproj. The module is the +leaf extracted by Gitea #512 so the runtime driver and the browse session assemble a broker +connection through **one** copy of the TLS posture, the PEM CA-pin chain validator, the credential +handling and the protocol-version mapping — a second copy would be a silent security divergence +(browse quietly not honouring a pinned CA while the driver does). Nothing under `bin/` or `obj/` +was reviewed. + +#### Notes on the categories that produced nothing + +Recorded rather than left implicit, because in a 275-line file "no issues" is a claim that should +be checkable: + +- **Concurrency (3).** The type is static and pure; the only mutable state is the + `Lazy` captured per validator, built with + `LazyThreadSafetyMode.ExecutionAndPublication`, and one is created per `BuildClientOptions` call + — i.e. per connect attempt — which is what makes a rotated CA file get picked up on the next + connect rather than pinned for process life. +- **Error handling (4).** Every failure on the pin path fails **closed**: an unreadable or empty PEM + returns null and the validator refuses; a `CryptographicException` out of `X509Chain.Build` is + caught and refused rather than escaping into the TLS callback as an opaque handshake crash; a + non-chain `SslPolicyErrors` bit (hostname mismatch, absent certificate) stays fatal, so the pin + replaces the trust *anchor* only. +- **Performance (6).** One chain build per handshake, `owned?.Dispose()` in a `finally` for the + copy made when the presented certificate is not already an `X509Certificate2`. No I/O on the + build path at all, which is the property that lets a config-validation caller use it. +- **Testing (9).** There is no `Driver.Mqtt.Transport.Tests` project; `BuildClientOptions` is + covered from `Driver.Mqtt.Tests` (`MqttConnectionTests` drives the real validator against real + self-signed chains — pinned CA accepted, stranger rejected, missing PEM fails closed, no I/O at + build time — and `MqttBrowseSessionTests` covers the client-id suffix and forced clean session). + That is genuine coverage of this module's behaviour through its two consumers, which is the shape + the extraction was for; not recorded as a finding. +- **Conventions / design (2, 7, 8).** Namespace `ZB.MOM.WW.OtOpcUa.Driver.Mqtt` (deliberately the + same as `.Contracts` and `.Driver`, so the extraction was not a source-breaking change), .NET 10, + `TreatWarningsAsErrors`, no MQTTnet leak into `.Contracts`. The csproj comment states the + module's charter — connection *policy* only, nothing that connects — and the file honours it. + +## Findings + +### Driver.Mqtt.Transport-001 + +| Field | Value | +|---|---| +| Severity | Medium | +| Category | Security | +| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs:120-128` (`ConfigureTls`) | +| Status | Resolved | + +**Description:** `AllowUntrustedServerCertificate` and `CaCertificatePath` are independent settings, +and when both are set the hatch wins **silently**: the branch installs +`WithAllowUntrustedCertificates(true)` + `WithIgnoreCertificateChainErrors(true)` + +`static _ => true` and returns before the pin is ever consulted. Every broker certificate is +accepted, pinned chain or not, and nothing in any log says so. + +The combination is not exotic. `allowUntrustedServerCertificate` is the dev/on-prem escape hatch an +operator turns on while standing a broker up, and `caCertificatePath` is what they add afterwards to +make MITM fail closed; forgetting to clear the first leaves a deployment that *looks* pinned in its +config, passes review, and validates nothing. The failure is silent in the direction that matters — +the operator's belief is "the pin is in force" and the reality is "any certificate is accepted" — +and neither the driver's health surface nor Test-connect can distinguish the two. + +Both knobs are explicit, so the fix is not to second-guess which one the operator meant: an +implementation that let the pin override the flag would break a deployment that genuinely wants the +hatch, and one that refused to build would fail a driver at connect over a config it has been +accepting. Saying so, once, at the seam that resolves the conflict is the proportionate answer. + +**Recommendation:** Warn (naming the pin path that is not being enforced) when +`AllowUntrustedServerCertificate` is set and `CaCertificatePath` is non-blank, and leave the flag +authoritative. The factory already takes an optional `ILogger` for exactly this class of deferred +diagnostic, and never formats credentials. + +**Resolution:** Fixed `74b85271`, 2026-08-10 — `ConfigureTls` logs a warning naming the shadowed CA +path before taking the accept-everything branch; behaviour is otherwise unchanged. Verified no +existing assertion depends on the branch being silent (713 tests green). + +### Driver.Mqtt.Transport-002 + +| Field | Value | +|---|---| +| Severity | Low | +| Category | Correctness & logic bugs | +| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs:97-100` | +| Status | Open | + +**Description:** Credentials are attached only when `Username` is non-empty: +`if (!string.IsNullOrEmpty(options.Username)) builder.WithCredentials(options.Username, options.Password ?? "")`. +A config carrying a `Password` and **no** `Username` therefore connects anonymously, with the +configured secret silently dropped. On MQTT 3.1.1 that is protocol-correct (a password may not be +sent without a username), but this driver defaults to `ProtocolVersion = V500`, where a +password-only CONNECT is legal and is how some brokers carry a token. + +Impact is bounded — the broker either accepts the anonymous connect (in which case the deployment is +less authenticated than its config says) or refuses it, which now surfaces as a classified CONNACK +rejection rather than a mystery. It is recorded because the silent-drop direction is the one an +operator cannot see from the config. + +**Recommendation:** Warn when `Password` is non-empty and `Username` is blank, so the discarded +credential is visible; optionally send password-only credentials under `V500`, where the protocol +permits it. Do **not** log the password itself — nothing on this path formats it today, and the +options record's `PrintMembers` override exists to keep it that way. + +**Resolution:** _(open)_ 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. diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index 1b08ba9c..0876c55a 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -404,6 +404,17 @@ nothing forever, so `OnReconnectedAsync` **throws** when zero filters are grante down rather than swallowing it. A SUBACK failure degrades just the affected RawPaths to `BadCommunicationError`. +There are **two** kinds of reconnect and only one of them fires that callback. A drop the connection's +own supervisor recovers keeps the same `MqttConnection`, so `Reconnected` → `OnReconnectedAsync` +re-establishes every live filter. A **session rebuild** — `ShutdownAsync` + `InitializeAsync` (the #527 +comms-loss watchdog's retry path, and an `ApplyDelta` that changes the endpoint or credentials) — builds +a *new* connection and re-attaches the *same* `MqttSubscriptionManager`, and the re-subscribe then comes +from the host (`DriverInstanceActor.ResubscribeDesired`) rather than from the callback. `AttachTo` +therefore **forgets the filters the previous session had established**: the dedupe set is a claim about +one session, and carrying it across made the host's re-subscribe resolve an empty pending set, send no +SUBSCRIBE at all, and leave the driver Connected/Healthy and permanently deaf. Sparkplug's +`SparkplugIngestor.AttachTo` resets its own session state for the same reason. + ## Write-through `IWritable` landed with [#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508). Before it, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs index 8b0dcd69..0f7e4333 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs @@ -119,6 +119,21 @@ public static class MqttClientOptionsFactory if (options.AllowUntrustedServerCertificate) { + // A configured CA pin is DEAD in this branch — every certificate is accepted, pinned chain + // or not. Silence here reads to an operator as "the pin is in force", which is the one + // belief this combination must never leave them holding: the pin is what they set to make a + // MITM fail closed, and the escape hatch is what makes it fail open. Warn, do not + // second-guess — the explicit flag stays authoritative, so a deployment that genuinely wants + // the hatch keeps working and merely says so in its log. + if (!string.IsNullOrWhiteSpace(options.CaCertificatePath)) + { + logger?.LogWarning( + "MQTT: allowUntrustedServerCertificate is on, so the pinned CA '{CaCertificatePath}' is NOT " + + "being enforced — ANY broker certificate is accepted. Clear the flag to make the pin " + + "effective.", + options.CaCertificatePath); + } + // Dev / on-prem escape hatch, off unless explicitly enabled — mirrors the // ServerHistorian TLS knobs. This is the ONLY branch that accepts a bad certificate. tls.WithAllowUntrustedCertificates(true) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs index 57219b83..78dbdf1f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs @@ -1,7 +1,6 @@ +// System.Buffers is load-bearing despite no type here naming it: OnApplicationMessageReceivedAsync +// calls ReadOnlySequence.ToArray(), which is BuffersExtensions', not a member. using System.Buffers; -using System.Net.Security; -using System.Security.Cryptography; -using System.Security.Cryptography.X509Certificates; using Microsoft.Extensions.Logging; using MQTTnet; using MQTTnet.Protocol; @@ -238,13 +237,14 @@ public sealed class MqttConnectRejectedException : Exception /// client it created and reports . /// /// -/// The publish leg is deliberately narrow. exists to satisfy -/// — whose entire production caller is -/// — not to make this a general MQTT publisher. This -/// driver has no IWritable leg in v1, so a Sparkplug rebirth NCMD is the only -/// outbound application message it ever sends. Sparkplug rebirth-on-reconnect itself is decided -/// by hanging off ; this type -/// only carries the bytes. +/// The publish leg is deliberately narrow. exists to carry +/// outbound application messages for callers that decide what to send, not to make this a +/// general MQTT publisher: (implemented for +/// 's NCMDs), MqttDriver.WriteAsync's encoded +/// write (Gitea #508 — the statement that this driver has no IWritable leg was true only +/// before it) and 's retained STATE. Sparkplug +/// rebirth-on-reconnect itself is decided by hanging +/// off ; this type only carries the bytes. /// /// public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, Sparkplug.IMqttPublishTransport diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs index ab4d3758..306ea3c8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs @@ -104,6 +104,18 @@ public sealed class MqttDriver /// How often the host-connectivity poll samples . private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1); + /// + /// Stand-in for a non-positive authored connectTimeoutSeconds — see . + /// Matches 's own default. + /// + private const int DefaultConnectTimeoutSeconds = 15; + + /// + /// Stand-in for a negative authored keepAliveSeconds — see . + /// Matches 's own default. + /// + private const int DefaultKeepAliveSeconds = 30; + private readonly string _driverInstanceId; private readonly ILogger? _logger; @@ -221,13 +233,13 @@ public sealed class MqttDriver ArgumentNullException.ThrowIfNull(options); ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId); - _options = options; + _options = Normalize(options); _driverInstanceId = driverInstanceId; _logger = logger; _primaryRoleSource = primaryRoleSource ?? UnmanagedDriverPrimaryRoleSource.Instance; - _subscriptions = CreateSubscriptionManager(options); - _sparkplug = CreateSparkplugIngestor(options); - RegisterAuthoredTags(options); + _subscriptions = CreateSubscriptionManager(_options); + _sparkplug = CreateSparkplugIngestor(_options); + RegisterAuthoredTags(_options); } /// @@ -372,7 +384,12 @@ public sealed class MqttDriver /// public async Task ShutdownAsync(CancellationToken cancellationToken) { - var lastMessage = ReadHealth().LastSuccessfulRead; + // GetHealth(), not ReadHealth(): the live connection's LastMessageUtc is the real last-message + // instant, and the stored snapshot only ever carries what was true at the last initialize. Taking + // the stored one here throws away the evidence a shutdown-then-initialize cycle (the #527 + // watchdog's reconnect) is measured against, reporting a driver that was delivering a moment ago + // as one that has never delivered. + var lastMessage = GetHealth().LastSuccessfulRead; if (await _lifecycleGate.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds), cancellationToken) .ConfigureAwait(false)) { @@ -932,6 +949,17 @@ public sealed class MqttDriver WriteHealth(new DriverHealth(DriverState.Initializing, ReadHealth().LastSuccessfulRead, null)); try { + // Tear down whatever this driver already holds BEFORE building a new session. Both fields + // published below (_connection, _primaryHost) are IAsyncDisposable and were previously + // overwritten, so a second initialize on a live driver stranded a connected, supervised + // broker session — plus a whole second primary-host connection — that no later dispose + // could reach. Reachable without any teardown of ours: DriverInstanceActor re-runs + // InitializeAsync with no ShutdownAsync when an ApplyDelta lands while the driver is in + // Connecting/Reconnecting (AdoptConfigDuringInit). Idempotent, so the ReinitializeAsync + // rebuild branch calling TeardownAsync first costs a no-op, and a first initialize sees + // three nulls. + await TeardownAsync().ConfigureAwait(false); + // A blank / unusable blob keeps the constructor's options rather than degrading the // driver to defaults pointing at localhost. var parsed = ParseOptions(driverConfigJson); @@ -1030,7 +1058,8 @@ public sealed class MqttDriver // deliberately: enums must round-trip by NAME. A second, divergent // JsonSerializerOptions is this repo's documented systemic enum bug, where an // AdminUI-authored config with a string enum field faults the driver. - return JsonSerializer.Deserialize(driverConfigJson, MqttJson.Options); + var parsed = JsonSerializer.Deserialize(driverConfigJson, MqttJson.Options); + return parsed is null ? null : Normalize(parsed); } catch (JsonException ex) { @@ -1050,6 +1079,54 @@ public sealed class MqttDriver } } + /// + /// Replaces out-of-range duration knobs with their documented defaults, once, at the single seam + /// every options instance this driver holds passes through (the constructor and + /// ). + /// + /// + /// + /// carries [Range] annotations, but nothing validates + /// them on the runtime path: the blob is deserialized straight out of the deploy artifact, so + /// a config authored (or hand-edited) with connectTimeoutSeconds: 0 or a negative value + /// reaches the driver unchecked. Both are load-bearing, and neither fails where an operator + /// would look: + /// + /// + /// + /// zero makes every new CancellationTokenSource(TimeSpan.Zero) in the connect, + /// subscribe and publish paths fire immediately, so the driver can never connect and + /// reports a timeout against a perfectly healthy broker; + /// + /// + /// a negative value is outside 's and + /// 's legal range, so it throws + /// out of and + /// — the two paths whose whole job is to not fail. + /// + /// + /// + /// Normalising here rather than at each use site keeps one answer for the whole driver: the + /// value also rides into , and + /// MqttClientOptionsFactory through . MaxPayloadBytes and + /// the reconnect backoffs are deliberately left alone — the manager's constructor and + /// already clamp those at their own use site. + /// + /// + private static MqttDriverOptions Normalize(MqttDriverOptions options) + { + var connectTimeout = options.ConnectTimeoutSeconds > 0 + ? options.ConnectTimeoutSeconds + : DefaultConnectTimeoutSeconds; + + // Zero keep-alive is legal MQTT ("no keep-alive"); only a negative one is nonsense. + var keepAlive = options.KeepAliveSeconds >= 0 ? options.KeepAliveSeconds : DefaultKeepAliveSeconds; + + return connectTimeout == options.ConnectTimeoutSeconds && keepAlive == options.KeepAliveSeconds + ? options + : options with { ConnectTimeoutSeconds = connectTimeout, KeepAliveSeconds = keepAlive }; + } + /// /// Adopts new options, rebuilding the subscription manager only when a setting it captures at /// construction actually changed. The rebuilt manager inherits the driver-owned diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs index 8438f456..14dc9d1d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -183,17 +183,35 @@ public sealed class MqttSubscriptionManager /// transport. /// /// - /// The manager does not own the connection's lifetime and never detaches — the driver disposes - /// both together. Call once per connection, and before the first - /// : a subscribe issued with no transport attached records its - /// intent (and warns) but sends nothing to a broker, and would only be established if the - /// connection later dropped and recovered. + /// + /// The manager does not own the connection's lifetime and never detaches — the driver disposes + /// both together. Call once per connection, and before the first + /// : a subscribe issued with no transport attached records its + /// intent (and warns) but sends nothing to a broker, and would only be established if the + /// connection later dropped and recovered. + /// + /// + /// Binding to a session forgets the previous session's established filters — the same + /// reset performs, and for the same reason: + /// this manager outlives a session rebuild. MqttDriver.AdoptOptions only + /// replaces it when the ingest identity changed, so a driver-host reconnect (ShutdownAsync + + /// InitializeAsync, #527's watchdog path) and a ReinitializeAsync that changes only the + /// endpoint/credentials both re-attach this instance to a brand-new broker session. + /// _subscribedTopics is a claim about the session that has just gone: carried across, + /// the host's post-reconnect re-subscribe resolves an empty pending set, issues no SUBSCRIBE + /// at all, and leaves a driver that reports Connected/Healthy and receives nothing until the + /// new session happens to drop — the silent-death outcome every other path in this + /// type is written to prevent. + /// /// /// The connection to observe. public void AttachTo(MqttConnection connection) { ArgumentNullException.ThrowIfNull(connection); + // FIRST, before delivery is wired: a new session holds none of the old one's subscriptions. + _subscribedTopics.Clear(); + _transport = connection; connection.MessageReceived += HandleMessage; connection.Reconnected += OnReconnectedAsync; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs index d614f05b..b43dd7bf 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs @@ -285,11 +285,19 @@ public sealed class SequenceTracker /// session token. /// /// - /// The reconnect seam (§3.6 invariant #5 — late join). After a dropped connection the driver - /// has missed an unknown number of messages, so the baseline it was holding is no longer - /// evidence of anything: carrying it across would let the stream look contiguous purely - /// because the edge node happened to publish the right number of messages while the driver was - /// away. + /// + /// The rationale is §3.6 invariant #5 (late join): after a dropped connection the driver has + /// missed an unknown number of messages, so the baseline it was holding is no longer evidence + /// of anything — carrying it across would let the stream look contiguous purely because the + /// edge node happened to publish the right number of messages while the driver was away. + /// + /// + /// The reconnect path does not call this, despite the reasoning being its. + /// SparkplugIngestor.ResetSessionState drops the whole per-node tracker map + /// instead, which additionally releases the entry for a node the plant has stopped + /// publishing. This member is the same reset for a single tracker, kept for a caller holding + /// one directly (the unit suite is currently the only one). + /// /// public void Reset() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs new file mode 100644 index 00000000..a20c1fca --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs @@ -0,0 +1,133 @@ +using System.Diagnostics; +using System.Text.Json; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// The driver's own session lifecycle, against the fake: what +/// does to the session it is replacing, and what a +/// mis-authored duration knob does to the two paths whose whole job is to not fail +/// ( / ). +/// +/// +/// Plain mode deliberately: it is the only mode whose initialize needs no SUBSCRIBE, and +/// answers CONNECT and PINGREQ and nothing else — so a Sparkplug driver +/// would park on its group SUBSCRIBE until the deadline and prove nothing about connections. +/// +[Trait("Category", "Unit")] +public sealed class MqttDriverSessionLifecycleTests +{ + /// + /// Driver.Mqtt-002. InitializeCoreAsync publishes _connection — an + /// — so re-initializing a live driver without tearing down first + /// stranded a connected, supervised broker session that no later dispose could reach, and the + /// abandoned session's reconnect supervisor kept it alive against the broker indefinitely. + /// + /// + /// + /// Reachable with no teardown of the driver's own: DriverInstanceActor re-runs + /// InitializeAsync with no ShutdownAsync when an ApplyDelta arrives while + /// the driver is Connecting/Reconnecting (AdoptConfigDuringInit). The + /// teardownFirst flag on the #527 retry path covers only the retry path. + /// + /// + /// The assertion is the broker's live socket count, not the driver's own view: a leak + /// is precisely the case where the driver believes it holds one session and the broker is + /// serving two. is asserted alongside it as the + /// control — without it, a second initialize that never dialled at all would satisfy + /// "exactly one live connection" just as well. + /// + /// + [Fact] + public async Task InitializeAsync_OnALiveDriver_TearsDownThePreviousSession() + { + using var broker = new MiniBroker(); + var options = BrokerOptions(broker.Port); + await using var driver = new MqttDriver(options, "d", null); + + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + await WaitForAsync(() => broker.LiveConnections == 1, "the first session to be established"); + + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken); + + broker.AcceptedCount.ShouldBe(2, "the second initialize must genuinely open a new session"); + await WaitForAsync( + () => broker.LiveConnections == 1, + "the replaced session to be torn down rather than left live and unreachable"); + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + /// + /// Driver.Mqtt-003. connectTimeoutSeconds carries a [Range] annotation that + /// nothing validates on the runtime path — the blob is deserialized straight out of the deploy + /// artifact. A negative value is outside the legal range of both + /// and , so + /// it threw out of shutdown and dispose — the two calls + /// that must always complete, and the ones a host makes while tearing a bad deployment down. + /// + [Fact] + public async Task ShutdownAndDispose_WithANegativeConnectTimeout_DoNotThrow() + { + var options = BrokerOptions(1) with { ConnectTimeoutSeconds = -1, KeepAliveSeconds = -1 }; + var driver = new MqttDriver(options, "d", null); + + await Should.NotThrowAsync(async () => + await driver.ShutdownAsync(TestContext.Current.CancellationToken)); + await Should.NotThrowAsync(async () => await driver.DisposeAsync()); + } + + /// + /// The control for the test above: the normalization must not quietly rewrite a legal + /// value, or a deployment's authored connect budget would silently become the default. + /// + [Fact] + public async Task InitializeAsync_WithAnAuthoredConnectTimeout_KeepsIt() + { + using var broker = new MiniBroker { Blackhole = true }; + var options = BrokerOptions(broker.Port) with { ConnectTimeoutSeconds = 2 }; + await using var driver = new MqttDriver(options, "d", null); + + var sw = Stopwatch.StartNew(); + await Should.ThrowAsync(async () => + await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken)); + sw.Stop(); + + sw.Elapsed.ShouldBeLessThan( + TimeSpan.FromSeconds(10), + "the authored 2s budget must still bound the connect, not the 15s default"); + } + + private static async Task WaitForAsync(Func condition, string what) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + while (DateTime.UtcNow < deadline) + { + if (condition()) + { + return; + } + + await Task.Delay(25, TestContext.Current.CancellationToken); + } + + throw new Xunit.Sdk.XunitException($"Timed out waiting for {what}."); + } + + private static MqttDriverOptions BrokerOptions(int port) => new() + { + Mode = MqttMode.Plain, + Host = "127.0.0.1", + Port = port, + UseTls = false, + ProtocolVersion = MqttProtocolVersion.V311, + KeepAliveSeconds = 300, + ConnectTimeoutSeconds = 5, + ReconnectMinBackoffSeconds = 1, + ReconnectMaxBackoffSeconds = 2, + }; + + private static string Json(MqttDriverOptions options) => JsonSerializer.Serialize(options, MqttJson.Options); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs index 5fc1e966..354caf4c 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs @@ -827,6 +827,66 @@ public sealed class MqttSubscriptionManagerTests transport.Calls.Count.ShouldBe(1); // the initial subscribe only — no zombie re-subscribe } + // --------------------------------------------------------------------------------- + // Session rebind — this manager OUTLIVES the broker session it established filters on + // --------------------------------------------------------------------------------- + + /// + /// Driver.Mqtt-001. A driver-host reconnect (ShutdownAsync + InitializeAsync, the #527 + /// watchdog's path) builds a brand-new but keeps this manager + /// — MqttDriver.AdoptOptions only replaces it when the ingest identity changed. The new + /// session holds none of the old one's subscriptions, so the host's post-reconnect re-subscribe + /// must actually reach the broker. While _subscribedTopics survived the rebind it resolved + /// an empty pending set, sent no SUBSCRIBE at all, and left the driver Connected/Healthy and + /// permanently deaf. + /// + /// + /// The rebound connection is deliberately a real, unconnected one: its + /// SubscribeAsync refuses ("no established session"), which EstablishAsync catches + /// and turns into BadCommunicationError on the affected references. That Bad IS the proof + /// the subscribe was attempted — with the bug the tag keeps the Good value it was last delivered, + /// because nothing is attempted at all. + /// + [Fact] + public async Task AttachTo_ForgetsTheFiltersEstablishedOnThePreviousSession() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + mgr.HandleMessage(OvenTopic, "21.5"u8.ToArray(), retained: false); + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good); + + // The session underneath is rebuilt and this manager is re-attached to it. + await using var rebound = new MqttConnection( + new MqttDriverOptions { Host = "127.0.0.1", Port = 1, UseTls = false }, "mqtt-1"); + mgr.AttachTo(rebound); + + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe( + BadCommunicationError, + "the re-subscribe must be issued against the NEW session, not skipped as already-established"); + } + + /// + /// The falsifiability control for the test above: within ONE session the dedupe must still hold, + /// or the fix would simply be "re-SUBSCRIBE everything on every subscribe" — which re-triggers + /// every topic's retained seed and is the behaviour _subscribedTopics exists to prevent. + /// + [Fact] + public async Task SubscribeAsync_TwiceOnTheSameSession_StillIssuesOneSubscribe() + { + var transport = new FakeTransport(); + var mgr = NewManager(transport); + mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]); + + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); + + transport.Calls.Count.ShouldBe(1); + } + /// Records every subscribe call so dedupe / re-subscribe can be asserted without a broker. private sealed class FakeTransport : IMqttSubscribeTransport {