merge: MQTT driver+transport review findings resolved (Driver.Mqtt-006/-007, Driver.Mqtt.Transport-002)
This commit is contained in:
Joseph Doherty
@@ -7,7 +7,7 @@
|
||||
| Review date | 2026-08-10 |
|
||||
| Commit reviewed | `b990adae` |
|
||||
| Status | Reviewed |
|
||||
| Open findings | 1 |
|
||||
| Open findings | 0 |
|
||||
|
||||
## Checklist coverage
|
||||
|
||||
@@ -110,7 +110,7 @@ existing assertion depends on the branch being silent (713 tests green).
|
||||
| Severity | Low |
|
||||
| Category | Correctness & logic bugs |
|
||||
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs:97-100` |
|
||||
| Status | Open |
|
||||
| Status | Resolved |
|
||||
|
||||
**Description:** Credentials are attached only when `Username` is non-empty:
|
||||
`if (!string.IsNullOrEmpty(options.Username)) builder.WithCredentials(options.Username, options.Password ?? "")`.
|
||||
@@ -129,4 +129,15 @@ credential is visible; optionally send password-only credentials under `V500`, w
|
||||
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)_
|
||||
**Resolution (2026-08-10):** Fixed `5d4dd17d` — the recommendation's first half only. A password
|
||||
configured with a blank username now logs a warning saying the connection is being made
|
||||
**anonymously** and naming the two ways to make that intentional; the credential is still not sent,
|
||||
and the secret is never formatted into the message. The optional half (a password-only CONNECT
|
||||
under `V500`) was deliberately **not** taken: 3.1.1 forbids it, so it cannot be the uniform
|
||||
behaviour of a factory that serves both versions, and adding it under `V500` alone would change the
|
||||
wire behaviour of deployments that connect happily today — the same "the explicit config stays
|
||||
authoritative, say so rather than second-guess it" line Driver.Mqtt.Transport-001 settled on.
|
||||
TDD: the warning assertion failed RED (`log.CountContaining("password") should be 1 but was 0`)
|
||||
before the change. Paired with a falsifying control — a config carrying **neither** credential must
|
||||
stay silent, or the fix would warn on every ordinary anonymous deployment and the first test would
|
||||
still pass. Suite green (734).
|
||||
|
||||
@@ -7,7 +7,7 @@
|
||||
| Review date | 2026-08-10 |
|
||||
| Commit reviewed | `b990adae` |
|
||||
| Status | Reviewed |
|
||||
| Open findings | 2 |
|
||||
| Open findings | 0 |
|
||||
|
||||
## Checklist coverage
|
||||
|
||||
@@ -214,7 +214,7 @@ with a comment naming the extension method that needs it. Build verified.
|
||||
| Severity | Low |
|
||||
| Category | Concurrency & thread safety |
|
||||
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs:278` |
|
||||
| Status | Open |
|
||||
| Status | Resolved |
|
||||
|
||||
**Description:** `DisposeAsync` calls `_gate.Dispose()`. `StopAsync` cancels the reconcile loop but
|
||||
does not await it, so a `ReconcileAsync` already in flight can be sitting in
|
||||
@@ -236,7 +236,16 @@ stated rationale — the gate holds no timer and no surviving registration), or
|
||||
the reconcile loop before dispose returns. The former is the smaller change and the one already
|
||||
argued for in this module.
|
||||
|
||||
**Resolution:** _(open)_
|
||||
**Resolution (2026-08-10):** Fixed `3f85ba98` — the first option. `DisposeAsync` no longer disposes
|
||||
`_gate`, and both the field and the dispose site now carry the rule and the reason, so the module
|
||||
states **one** answer for its lifecycle primitives rather than two. Nothing is leaked (the gate
|
||||
holds no timer and no registration) and `ClaimAsync` still re-checks `_disposed` under it, so no
|
||||
claim can be created after dispose either way. TDD: a test that disposes while a reconcile is in
|
||||
flight failed RED with `log.Captured should not contain an element satisfying (name ==
|
||||
"ObjectDisposedException") but does`. It is deterministic rather than timing-dependent — the gate is
|
||||
taken *before* the connect starts, so a `MiniBroker` in blackhole mode (TCP accepted, CONNECT never
|
||||
answered) having accepted the socket proves the claim is inside the gate when dispose runs. Suite
|
||||
green (734).
|
||||
|
||||
### Driver.Mqtt-007
|
||||
|
||||
@@ -245,7 +254,7 @@ argued for in this module.
|
||||
| Severity | Low |
|
||||
| Category | Error handling & resilience |
|
||||
| Location | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs:406-424` (`OnReconnectedAsync`) |
|
||||
| Status | Open |
|
||||
| Status | Resolved |
|
||||
|
||||
**Description:** When the reconnect path finds live subscribed topics but resolves **zero** filters
|
||||
for them, it logs a warning and returns normally. The type's own contract, stated two paragraphs
|
||||
@@ -263,7 +272,19 @@ which is an argument for leaving the code alone, not for leaving the contract am
|
||||
*this* deaf-reconnect shape is exempt from the type's own rule. Do not leave the two answers side by
|
||||
side unexplained.
|
||||
|
||||
**Resolution:** _(open)_
|
||||
**Resolution (2026-08-10):** Fixed `2298485b` — it throws, and the remarks now state the single rule.
|
||||
Exempting it was considered and rejected: the reason the partial-rejection case is exempt is that
|
||||
tearing the session down would punish the tags still flowing, and here there are none — every live
|
||||
topic is in the set that resolved to nothing — so the teardown costs nothing and the retry is not
|
||||
futile, since it runs against whatever `Register` has published by then. The `subscribed.Count == 0`
|
||||
branch stays a debug no-op: a reconnect with nothing subscribed yet re-establishes nothing because
|
||||
there is nothing to establish. TDD: `Should.ThrowAsync<InvalidOperationException>` failed RED
|
||||
(`should throw System.InvalidOperationException but did not`). The test reaches the state the only
|
||||
way production can — `Register` prunes `_subscribedTopics` of the topics its new table dropped, but a
|
||||
SUBSCRIBE already in flight records its own topic *after* that prune, so a deploy landing inside the
|
||||
round-trip leaves a live topic no authored tag resolves to — driven through a `DuringSubscribe` hook
|
||||
on the existing fake transport. Paired with the control that a reconnect with no live subscriptions
|
||||
stays a no-op, which a blanket `filters.Count == 0` throw would have broken. Suite green (734).
|
||||
|
||||
### Driver.Mqtt-008
|
||||
|
||||
|
||||
@@ -98,6 +98,18 @@ public static class MqttClientOptionsFactory
|
||||
{
|
||||
builder = builder.WithCredentials(options.Username, options.Password ?? string.Empty);
|
||||
}
|
||||
else if (!string.IsNullOrEmpty(options.Password))
|
||||
{
|
||||
// A password with no username is dropped, and the drop is the direction an operator cannot
|
||||
// see: the config reads as authenticated and the CONNECT goes out anonymous. MQTT 3.1.1 may
|
||||
// not carry a password without a username, so synthesizing one is not available on every
|
||||
// protocol version this factory serves — and inventing a password-only CONNECT under 5.0
|
||||
// only would change the wire behaviour of deployments that connect happily today. Say so
|
||||
// once, here, and leave the configured values authoritative. Never format the secret.
|
||||
logger?.LogWarning(
|
||||
"MQTT: a password is configured with no username, so it is NOT being sent — this connection "
|
||||
+ "will be made anonymously. Set the username, or clear the password to make that intentional.");
|
||||
}
|
||||
|
||||
builder = builder.WithTlsOptions(tls => ConfigureTls(tls, options, logger));
|
||||
|
||||
|
||||
@@ -397,6 +397,15 @@ public sealed class MqttSubscriptionManager
|
||||
/// rest keep flowing. That is the plan's "SUBACK failure → per-ref Bad, not hang", scoped
|
||||
/// to the case where it does not cost the whole driver.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Live topics resolving to zero filters is a total failure too</b> — it re-establishes
|
||||
/// nothing, so it throws by the same rule rather than warning and returning. It costs no
|
||||
/// other tag (every live topic is in the set that resolved to nothing) and it is not
|
||||
/// permanent, because the retry runs against whatever <see cref="Register"/> has published
|
||||
/// by then. It is reachable only when a deploy lands inside a subscribe round-trip:
|
||||
/// <see cref="Register"/> prunes <c>_subscribedTopics</c> of the topics its new table no
|
||||
/// longer carries, but an in-flight SUBSCRIBE records its own topic after that prune ran.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="cancellationToken">The connection's lifetime token; cancelled by its dispose.</param>
|
||||
/// <returns>A task that completes when the re-subscribe has been answered.</returns>
|
||||
@@ -421,24 +430,21 @@ public sealed class MqttSubscriptionManager
|
||||
var filters = BuildFilters(live, _defaultQos);
|
||||
if (filters.Count == 0)
|
||||
{
|
||||
// Silence here is how the wildcard defect hid: no exception, no log, no SUBSCRIBE. If any
|
||||
// topic was live, resolving zero filters for it is an anomaly and must be loud.
|
||||
// Silence here is how the wildcard defect hid: no exception, no log, no SUBSCRIBE. Live
|
||||
// topics resolving to zero filters re-establishes nothing, which is the total-failure case
|
||||
// by a different route, so it gets the total-failure answer — see the remarks.
|
||||
if (subscribed.Count > 0)
|
||||
{
|
||||
_logger?.LogWarning(
|
||||
"MQTT driver '{DriverId}': reconnect resolved NO filters for {TopicCount} live topic(s) "
|
||||
+ "({Topics}); those subscriptions are not being re-established.",
|
||||
_driverId,
|
||||
subscribed.Count,
|
||||
string.Join(", ", subscribed));
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger?.LogDebug(
|
||||
"MQTT driver '{DriverId}': reconnect had no live subscriptions to re-establish.",
|
||||
_driverId);
|
||||
throw new InvalidOperationException(
|
||||
$"MQTT driver '{_driverId}': reconnect resolved NO filters for {subscribed.Count} live "
|
||||
+ $"topic(s) ({string.Join(", ", subscribed)}); tearing the session down rather than "
|
||||
+ "serving a connection that receives nothing.");
|
||||
}
|
||||
|
||||
_logger?.LogDebug(
|
||||
"MQTT driver '{DriverId}': reconnect had no live subscriptions to re-establish.",
|
||||
_driverId);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
@@ -73,7 +73,10 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable
|
||||
private readonly ILogger? _logger;
|
||||
private readonly TimeProvider _timeProvider;
|
||||
|
||||
/// <summary>Serializes claim/release against each other and against <see cref="DisposeAsync"/>.</summary>
|
||||
/// <summary>
|
||||
/// Serializes claim/release against each other and against <see cref="DisposeAsync"/>. Never
|
||||
/// disposed — see <see cref="DisposeAsync"/> for why.
|
||||
/// </summary>
|
||||
private readonly SemaphoreSlim _gate = new(1, 1);
|
||||
|
||||
private CancellationTokenSource? _loopCts;
|
||||
@@ -275,7 +278,14 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable
|
||||
ex, "MQTT driver '{DriverId}': Sparkplug primary-host stand-down failed during dispose.", _driverId);
|
||||
}
|
||||
|
||||
_gate.Dispose();
|
||||
// _gate is deliberately NOT disposed, matching MqttConnection's rule for its own lifecycle
|
||||
// primitives. StopAsync cancels the reconcile loop without awaiting it, so a ReconcileAsync can
|
||||
// still be inside the gate — or on its way to the finally-release — when this returns. Against a
|
||||
// disposed semaphore that release throws an ObjectDisposedException naming SemaphoreSlim, which
|
||||
// is logged as a reconcile failure and tells an operator nothing (and hides whatever the
|
||||
// reconcile was really failing on). Left undisposed the late release is a harmless no-op: the
|
||||
// gate holds no timer and no registration, and ClaimAsync re-checks _disposed under it, so no
|
||||
// claim can be created after dispose either way.
|
||||
}
|
||||
|
||||
private async Task ReconcileLoopAsync(CancellationToken ct)
|
||||
|
||||
@@ -366,6 +366,43 @@ public sealed class MqttConnectionTests
|
||||
built.Credentials.GetUserName(built).ShouldBe("svc");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Driver.Mqtt.Transport-002.</b> A configured password with no username is dropped — MQTT
|
||||
/// 3.1.1 may not carry one without the other, and this factory does not synthesize the
|
||||
/// password-only CONNECT 5.0 would permit. What must not happen is the drop being silent: the
|
||||
/// deployment then believes it authenticates and does not, which is the one direction an
|
||||
/// operator cannot see from the config. The secret itself stays out of the log.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void BuildClientOptions_PasswordWithoutUsername_IsDroppedLoudlyAndWithoutLeakingIt()
|
||||
{
|
||||
const string secret = "s3cr3t-that-must-not-be-logged";
|
||||
var log = new CapturingLogger();
|
||||
|
||||
var built = MqttClientOptionsFactory.BuildClientOptions(
|
||||
new MqttDriverOptions { Host = Host, Password = secret }, clientIdSuffix: null, logger: log);
|
||||
|
||||
built.Credentials.ShouldBeNull();
|
||||
log.CountContaining("password").ShouldBe(1);
|
||||
log.CountContaining(secret).ShouldBe(0);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The falsifying control for the test above: an ordinary anonymous deployment configures
|
||||
/// neither credential and must stay silent, or every such driver warns on every connect.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void BuildClientOptions_NoCredentialsAtAll_SaysNothing()
|
||||
{
|
||||
var log = new CapturingLogger();
|
||||
|
||||
MqttClientOptionsFactory.BuildClientOptions(
|
||||
new MqttDriverOptions { Host = Host }, clientIdSuffix: null, logger: log)
|
||||
.Credentials.ShouldBeNull();
|
||||
|
||||
log.CountContaining("password").ShouldBe(0);
|
||||
}
|
||||
|
||||
/// <summary>The ctor must not touch the network — the universal-browser throwaway-instance pattern.</summary>
|
||||
[Fact]
|
||||
public void Ctor_IsConnectionFree()
|
||||
|
||||
@@ -583,6 +583,51 @@ public sealed class MqttSubscriptionManagerTests
|
||||
mgr.Values.Read(OvenTempPath).StatusCode.ShouldBe(Good);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Driver.Mqtt-007.</b> Live subscribed topics that resolve to ZERO filters re-establish
|
||||
/// nothing, which is the same outcome as a total SUBACK failure and must get the same answer:
|
||||
/// throw, so <see cref="MqttConnection"/> tears the fresh session down and retries rather than
|
||||
/// serving a connected-but-deaf driver. A warning was not enough — the whole point of the type's
|
||||
/// split contract is that "no subscription was re-established" is fatal to the session.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The state is reached the only way production can reach it: <see cref="MqttSubscriptionManager.Register"/>
|
||||
/// prunes <c>_subscribedTopics</c> of the topics its new table no longer carries, but a SUBSCRIBE
|
||||
/// already in flight records its topic <i>after</i> that prune has run — so a redeploy landing
|
||||
/// inside the subscribe round-trip leaves a live topic no authored tag resolves to.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task OnReconnectedAsync_LiveTopicResolvingNoFilter_Throws_RatherThanServingADeafSession()
|
||||
{
|
||||
var transport = new FakeTransport();
|
||||
var mgr = NewManager(transport);
|
||||
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
|
||||
|
||||
transport.DuringSubscribe = () => mgr.Register([]); // the redeploy lands mid-round-trip
|
||||
await mgr.SubscribeAsync([OvenTempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
||||
transport.DuringSubscribe = null;
|
||||
|
||||
await Should.ThrowAsync<InvalidOperationException>(
|
||||
() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The falsifying control: a reconnect with no live subscriptions at all re-establishes nothing
|
||||
/// because there is nothing to establish, and must stay a no-op. Throwing here would tear down
|
||||
/// the session of every driver whose tags have not been subscribed yet.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task OnReconnectedAsync_NoLiveSubscriptions_IsANoOp()
|
||||
{
|
||||
var transport = new FakeTransport();
|
||||
var mgr = NewManager(transport);
|
||||
mgr.Register([Tag(OvenTempPath, $$"""{"topic":"{{OvenTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""")]);
|
||||
|
||||
await Should.NotThrowAsync(() => mgr.OnReconnectedAsync(TestContext.Current.CancellationToken));
|
||||
|
||||
transport.Calls.Count.ShouldBe(0);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------
|
||||
// C1 — wildcard-authored tags on the paths that reconstruct state from a FILTER string.
|
||||
// Delivery matches an incoming topic (concrete); reconnect + degrade key off the subscribed
|
||||
@@ -898,10 +943,18 @@ public sealed class MqttSubscriptionManagerTests
|
||||
/// <summary>When set, the whole subscribe call throws instead of answering.</summary>
|
||||
public Exception? Throw { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Runs inside the subscribe round-trip, so a test can land a concurrent
|
||||
/// <see cref="MqttSubscriptionManager.Register"/> in the window between its prune and this
|
||||
/// call recording its topic as live.
|
||||
/// </summary>
|
||||
public Action? DuringSubscribe { get; set; }
|
||||
|
||||
public Task<IReadOnlyList<MqttSubscribeOutcome>> SubscribeAsync(
|
||||
IReadOnlyList<MqttTopicSubscription> filters, CancellationToken cancellationToken)
|
||||
{
|
||||
Calls.Add(filters);
|
||||
DuringSubscribe?.Invoke();
|
||||
if (Throw is not null) throw Throw;
|
||||
IReadOnlyList<MqttSubscribeOutcome> outcomes =
|
||||
[.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, !Reject(f.Topic), Reject(f.Topic) ? "NotAuthorized" : "GrantedQoS1"))];
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MQTTnet;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
@@ -254,10 +255,105 @@ public sealed class SparkplugPrimaryHostTests
|
||||
await Should.NotThrowAsync(async () => await host.DisposeAsync());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Driver.Mqtt-006.</b> Dispose runs while a reconcile is in flight — the claim is inside the
|
||||
/// gate, blocked on a broker that accepts TCP and never answers CONNECT — and must not leave that
|
||||
/// reconcile releasing a disposed semaphore. The symptom is an <c>ObjectDisposedException</c>
|
||||
/// naming <c>SemaphoreSlim</c>, logged as a reconcile failure: it tells an operator nothing about
|
||||
/// what happened, and it hides whatever the reconcile was actually failing on.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Deterministic rather than timing-dependent: the gate is taken before the connect starts, so
|
||||
/// the broker having accepted the socket proves the claim is inside it when dispose runs.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task Disposing_while_a_reconcile_is_in_flight_does_not_fault_on_a_disposed_gate()
|
||||
{
|
||||
using var broker = new MiniBroker { Blackhole = true };
|
||||
var log = new ExceptionCapturingLogger();
|
||||
var options = Options() with
|
||||
{
|
||||
Host = "127.0.0.1",
|
||||
Port = broker.Port,
|
||||
UseTls = false,
|
||||
ProtocolVersion = MqttProtocolVersion.V311,
|
||||
ConnectTimeoutSeconds = 2,
|
||||
};
|
||||
var host = new SparkplugPrimaryHost(options, "drv-1", new FixedRole(shouldActAsPrimary: true), log);
|
||||
|
||||
var starting = host.StartAsync(CancellationToken.None);
|
||||
(await WaitUntilAsync(() => broker.AcceptedCount > 0, TimeSpan.FromSeconds(10)))
|
||||
.ShouldBeTrue("the claim must be inside the gate before dispose runs");
|
||||
|
||||
await host.DisposeAsync();
|
||||
await starting;
|
||||
|
||||
log.Captured.ShouldNotContain(
|
||||
name => name == nameof(ObjectDisposedException),
|
||||
"a teardown racing an in-flight reconcile must not surface a disposed-semaphore failure");
|
||||
}
|
||||
|
||||
private static async Task<bool> WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
|
||||
{
|
||||
var deadline = DateTime.UtcNow + timeout;
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
if (condition())
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
await Task.Delay(25);
|
||||
}
|
||||
|
||||
return condition();
|
||||
}
|
||||
|
||||
private sealed class FixedRole(bool shouldActAsPrimary) : IDriverPrimaryRoleSource
|
||||
{
|
||||
public bool ShouldActAsPrimary { get; private set; } = shouldActAsPrimary;
|
||||
|
||||
public void PublishPrimaryRole(bool value) => ShouldActAsPrimary = value;
|
||||
}
|
||||
|
||||
/// <summary>Records the type name of every exception logged, which is the observable here.</summary>
|
||||
private sealed class ExceptionCapturingLogger : ILogger
|
||||
{
|
||||
private readonly List<string> _captured = [];
|
||||
|
||||
public IReadOnlyList<string> Captured
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_captured)
|
||||
{
|
||||
return [.. _captured];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public IDisposable? BeginScope<TState>(TState state)
|
||||
where TState : notnull
|
||||
=> null;
|
||||
|
||||
public bool IsEnabled(LogLevel logLevel) => true;
|
||||
|
||||
public void Log<TState>(
|
||||
LogLevel logLevel,
|
||||
EventId eventId,
|
||||
TState state,
|
||||
Exception? exception,
|
||||
Func<TState, Exception?, string> formatter)
|
||||
{
|
||||
if (exception is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
lock (_captured)
|
||||
{
|
||||
_captured.Add(exception.GetType().Name);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user