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); + } + } + } }