fix(mqtt): stop disposing the primary-host gate a live reconcile still holds
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".
This commit is contained in:
@@ -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