dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

merge: Mqtt runtime + Transport code review — deaf-reconnect subscription reset (High), teardown-before-reinit session leak (High), timeout normalization, dead CA-pin warning; 12 findings

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-10 07:02:15 -04:00
parent 147b624437 fefb240350
commit d68009106b
10 changed files with 820 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code-reviews/Driver.Mqtt.Transport/findings.md
+132
View File
@@ -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<X509Certificate2Collection?>` 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)_
code-reviews/Driver.Mqtt/findings.md
+340
View File
@@ -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<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 | 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.
docs/drivers/Mqtt.md
+11
View File
@@ -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,
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs
+15
View File
@@ -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)
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs
+10 -10
View File
@@ -1,7 +1,6 @@
// System.Buffers is load-bearing despite no type here naming it: OnApplicationMessageReceivedAsync
// calls ReadOnlySequence<byte>.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 <see cref="ObjectDisposedException"/>.
/// </para>
/// <para>
/// <b>The publish leg is deliberately narrow.</b> <see cref="PublishAsync"/> exists to satisfy
/// <see cref="Sparkplug.IMqttPublishTransport"/> — whose entire production caller is
/// <see cref="Sparkplug.RebirthRequester"/> — not to make this a general MQTT publisher. This
/// driver has no <c>IWritable</c> leg in v1, so a Sparkplug rebirth NCMD is the <i>only</i>
/// outbound application message it ever sends. Sparkplug rebirth-on-reconnect itself is decided
/// by <see cref="Sparkplug.SparkplugIngestor"/> hanging off <see cref="Reconnected"/>; this type
/// only carries the bytes.
/// <b>The publish leg is deliberately narrow.</b> <see cref="PublishAsync"/> exists to carry
/// outbound application messages for callers that decide <i>what</i> to send, not to make this a
/// general MQTT publisher: <see cref="Sparkplug.IMqttPublishTransport"/> (implemented for
/// <see cref="Sparkplug.RebirthRequester"/>'s NCMDs), <c>MqttDriver.WriteAsync</c>'s encoded
/// write (Gitea #508 — the statement that this driver has no <c>IWritable</c> leg was true only
/// before it) and <see cref="Sparkplug.SparkplugPrimaryHost"/>'s retained STATE. Sparkplug
/// rebirth-on-reconnect itself is decided by <see cref="Sparkplug.SparkplugIngestor"/> hanging
/// off <see cref="Reconnected"/>; this type only carries the bytes.
/// </para>
/// </remarks>
public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, Sparkplug.IMqttPublishTransport
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
+83 -6
View File
@@ -104,6 +104,18 @@ public sealed class MqttDriver
/// <summary>How often the host-connectivity poll samples <see cref="MqttConnection.State"/>.</summary>
private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1);
/// <summary>
/// Stand-in for a non-positive authored <c>connectTimeoutSeconds</c> — see <see cref="Normalize"/>.
/// Matches <see cref="MqttDriverOptions.ConnectTimeoutSeconds"/>'s own default.
/// </summary>
private const int DefaultConnectTimeoutSeconds = 15;
/// <summary>
/// Stand-in for a negative authored <c>keepAliveSeconds</c> — see <see cref="Normalize"/>.
/// Matches <see cref="MqttDriverOptions.KeepAliveSeconds"/>'s own default.
/// </summary>
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);
}
/// <inheritdoc />
@@ -372,7 +384,12 @@ public sealed class MqttDriver
/// <inheritdoc />
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<MqttDriverOptions>(driverConfigJson, MqttJson.Options);
var parsed = JsonSerializer.Deserialize<MqttDriverOptions>(driverConfigJson, MqttJson.Options);
return parsed is null ? null : Normalize(parsed);
}
catch (JsonException ex)
{
@@ -1050,6 +1079,54 @@ public sealed class MqttDriver
}
}
/// <summary>
/// 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
/// <see cref="ParseOptions"/>).
/// </summary>
/// <remarks>
/// <para>
/// <see cref="MqttDriverOptions"/> carries <c>[Range]</c> 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 <c>connectTimeoutSeconds: 0</c> or a negative value
/// reaches the driver unchecked. Both are load-bearing, and neither fails where an operator
/// would look:
/// </para>
/// <list type="bullet">
/// <item>
/// zero makes every <c>new CancellationTokenSource(TimeSpan.Zero)</c> in the connect,
/// subscribe and publish paths fire immediately, so the driver can never connect and
/// reports a timeout against a perfectly healthy broker;
/// </item>
/// <item>
/// a negative value is outside <see cref="SemaphoreSlim.WaitAsync(TimeSpan)"/>'s and
/// <see cref="CancellationTokenSource"/>'s legal range, so it throws
/// <see cref="ArgumentOutOfRangeException"/> out of <see cref="ShutdownAsync"/> and
/// <see cref="DisposeAsync"/> — the two paths whose whole job is to not fail.
/// </item>
/// </list>
/// <para>
/// Normalising here rather than at each use site keeps one answer for the whole driver: the
/// value also rides into <see cref="MqttConnection"/>, <see cref="SparkplugPrimaryHost"/> and
/// <c>MqttClientOptionsFactory</c> through <see cref="_options"/>. <c>MaxPayloadBytes</c> and
/// the reconnect backoffs are deliberately left alone — the manager's constructor and
/// <see cref="MqttConnection.NextBackoff"/> already clamp those at their own use site.
/// </para>
/// </remarks>
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 };
}
/// <summary>
/// Adopts new options, rebuilding the subscription manager only when a setting it captures at
/// construction actually changed. The rebuilt manager inherits the driver-owned
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs
+23 -5
View File
@@ -183,17 +183,35 @@ public sealed class MqttSubscriptionManager
/// transport.
/// </summary>
/// <remarks>
/// The manager does not own the connection's lifetime and never detaches — the driver disposes
/// both together. Call once per connection, and <b>before</b> the first
/// <see cref="SubscribeAsync"/>: 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.
/// <para>
/// The manager does not own the connection's lifetime and never detaches — the driver disposes
/// both together. Call once per connection, and <b>before</b> the first
/// <see cref="SubscribeAsync"/>: 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.
/// </para>
/// <para>
/// <b>Binding to a session forgets the previous session's established filters</b> — the same
/// reset <see cref="Sparkplug.SparkplugIngestor.AttachTo"/> performs, and for the same reason:
/// this manager <i>outlives</i> a session rebuild. <c>MqttDriver.AdoptOptions</c> only
/// replaces it when the ingest identity changed, so a driver-host reconnect (ShutdownAsync +
/// InitializeAsync, #527's watchdog path) and a <c>ReinitializeAsync</c> that changes only the
/// endpoint/credentials both re-attach <i>this</i> instance to a brand-new broker session.
/// <c>_subscribedTopics</c> 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
/// <i>new</i> session happens to drop — the silent-death outcome every other path in this
/// type is written to prevent.
/// </para>
/// </remarks>
/// <param name="connection">The connection to observe.</param>
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;
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SequenceTracker.cs
+13 -5
View File
@@ -285,11 +285,19 @@ public sealed class SequenceTracker
/// session token.
/// </summary>
/// <remarks>
/// 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.
/// <para>
/// 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.
/// </para>
/// <para>
/// <b>The reconnect path does not call this</b>, despite the reasoning being its.
/// <c>SparkplugIngestor.ResetSessionState</c> drops the whole per-node tracker <i>map</i>
/// 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).
/// </para>
/// </remarks>
public void Reset()
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs
+133
View File
@@ -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;
/// <summary>
/// The driver's own session lifecycle, against the <see cref="MiniBroker"/> fake: what
/// <see cref="MqttDriver.InitializeAsync"/> does to the session it is <i>replacing</i>, and what a
/// mis-authored duration knob does to the two paths whose whole job is to not fail
/// (<see cref="MqttDriver.ShutdownAsync"/> / <see cref="MqttDriver.DisposeAsync"/>).
/// </summary>
/// <remarks>
/// Plain mode deliberately: it is the only mode whose initialize needs no SUBSCRIBE, and
/// <see cref="MiniBroker"/> 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.
/// </remarks>
[Trait("Category", "Unit")]
public sealed class MqttDriverSessionLifecycleTests
{
/// <summary>
/// <b>Driver.Mqtt-002.</b> <c>InitializeCoreAsync</c> publishes <c>_connection</c> — an
/// <see cref="IAsyncDisposable"/> — 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.
/// </summary>
/// <remarks>
/// <para>
/// Reachable with no teardown of the driver's own: <c>DriverInstanceActor</c> re-runs
/// <c>InitializeAsync</c> with no <c>ShutdownAsync</c> when an <c>ApplyDelta</c> arrives while
/// the driver is Connecting/Reconnecting (<c>AdoptConfigDuringInit</c>). The
/// <c>teardownFirst</c> flag on the #527 retry path covers only the retry path.
/// </para>
/// <para>
/// The assertion is the broker's <b>live socket count</b>, 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. <see cref="MiniBroker.AcceptedCount"/> 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.
/// </para>
/// </remarks>
[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);
}
/// <summary>
/// <b>Driver.Mqtt-003.</b> <c>connectTimeoutSeconds</c> carries a <c>[Range]</c> 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
/// <see cref="SemaphoreSlim.WaitAsync(TimeSpan)"/> and <see cref="CancellationTokenSource"/>, so
/// it threw <see cref="ArgumentOutOfRangeException"/> out of shutdown and dispose — the two calls
/// that must always complete, and the ones a host makes while tearing a bad deployment down.
/// </summary>
[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());
}
/// <summary>
/// The control for the test above: the normalization must not quietly rewrite a <i>legal</i>
/// value, or a deployment's authored connect budget would silently become the default.
/// </summary>
[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<TimeoutException>(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<bool> 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);
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttSubscriptionManagerTests.cs
+60
View File
@@ -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
// ---------------------------------------------------------------------------------
/// <summary>
/// <b>Driver.Mqtt-001.</b> A driver-host reconnect (ShutdownAsync + InitializeAsync, the #527
/// watchdog's path) builds a brand-new <see cref="MqttConnection"/> but keeps <i>this</i> manager
/// — <c>MqttDriver.AdoptOptions</c> 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 <c>_subscribedTopics</c> survived the rebind it resolved
/// an empty pending set, sent no SUBSCRIBE at all, and left the driver Connected/Healthy and
/// permanently deaf.
/// </summary>
/// <remarks>
/// The rebound connection is deliberately a real, <b>unconnected</b> one: its
/// <c>SubscribeAsync</c> refuses ("no established session"), which <c>EstablishAsync</c> catches
/// and turns into <c>BadCommunicationError</c> 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.
/// </remarks>
[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");
}
/// <summary>
/// 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 <c>_subscribedTopics</c> exists to prevent.
/// </summary>
[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);
}
/// <summary>Records every subscribe call so dedupe / re-subscribe can be asserted without a broker.</summary>
private sealed class FakeTransport : IMqttSubscribeTransport
{