From 3f85ba985ecdf4f59d519ea0377ffefecedfa0b7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 10 Aug 2026 07:58:38 -0400 Subject: [PATCH] fix(mqtt): stop disposing the primary-host gate a live reconcile still holds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Driver.Mqtt-006 (Low, concurrency). DisposeAsync disposed _gate while StopAsync cancels the reconcile loop WITHOUT awaiting it, so a ReconcileAsync already in flight could be inside _gate.WaitAsync, or on its way to the finally-release, when the semaphore went away. The release then threw ObjectDisposedException naming SemaphoreSlim, logged as a reconcile failure — telling an operator nothing about what happened and hiding whatever the reconcile was really failing on. Left undisposed, matching MqttConnection's documented rule for its own lifecycle primitives and its stated rationale: a late callback racing teardown should be a benign no-op rather than an ObjectDisposedException naming the wrong type. Two lifecycle types in one module disagreeing about this is what gets copied into the third. 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. Test 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) accepting the socket proves the claim is inside the gate when dispose runs. RED: log.Captured contained "ObjectDisposedException". --- .../Sparkplug/SparkplugPrimaryHost.cs | 14 ++- .../SparkplugPrimaryHostTests.cs | 96 +++++++++++++++++++ 2 files changed, 108 insertions(+), 2 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs index 7911c872..02c1d696 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs @@ -73,7 +73,10 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable private readonly ILogger? _logger; private readonly TimeProvider _timeProvider; - /// Serializes claim/release against each other and against . + /// + /// Serializes claim/release against each other and against . Never + /// disposed — see for why. + /// 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) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs index 451a060c..0cffd5b5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs @@ -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()); } + /// + /// Driver.Mqtt-006. 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 ObjectDisposedException + /// naming SemaphoreSlim, logged as a reconcile failure: it tells an operator nothing about + /// what happened, and it hides whatever the reconcile was actually failing on. + /// + /// + /// 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. + /// + [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 WaitUntilAsync(Func 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; } + + /// Records the type name of every exception logged, which is the observable here. + private sealed class ExceptionCapturingLogger : ILogger + { + private readonly List _captured = []; + + public IReadOnlyList Captured + { + get + { + lock (_captured) + { + return [.. _captured]; + } + } + } + + public IDisposable? BeginScope(TState state) + where TState : notnull + => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + if (exception is null) + { + return; + } + + lock (_captured) + { + _captured.Add(exception.GetType().Name); + } + } + } }