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/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 {