fix(mqtt): back off the Sparkplug primary-host claim loop after failures (#602 STAB-27)
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
@@ -66,6 +66,14 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable
|
||||
/// <summary>How often the role gate is sampled. Bounds the failover republish latency.</summary>
|
||||
private static readonly TimeSpan RoleSampleInterval = TimeSpan.FromSeconds(1);
|
||||
|
||||
/// <summary>Ceiling on the delay between failed claim attempts (STAB-27).</summary>
|
||||
internal static readonly TimeSpan MaxClaimBackoff = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>Consecutive failed claim attempts; reset by a successful claim or a stand-down. Reconcile-loop-owned.</summary>
|
||||
private int _claimFailures;
|
||||
|
||||
private DateTimeOffset _nextClaimAttemptUtc;
|
||||
|
||||
private readonly MqttDriverOptions _options;
|
||||
private readonly string _hostId;
|
||||
private readonly string _driverId;
|
||||
@@ -321,10 +329,21 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable
|
||||
{
|
||||
if (_roleSource.ShouldActAsPrimary)
|
||||
{
|
||||
// STAB-27: a claim that keeps failing backs off instead of redialling (and logging a
|
||||
// Warning) every RoleSampleInterval for as long as the broker refuses it. The role is still
|
||||
// sampled each tick — only the connect attempt waits.
|
||||
if (_claimFailures > 0 && _timeProvider.GetUtcNow() < _nextClaimAttemptUtc)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await ClaimAsync(cancellationToken).ConfigureAwait(false);
|
||||
_claimFailures = 0;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Standing down is not a failure, and a later re-promotion should claim at once.
|
||||
_claimFailures = 0;
|
||||
await ReleaseAsync("this node is not the redundancy Primary", cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
@@ -334,14 +353,42 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger?.LogWarning(
|
||||
ex,
|
||||
"MQTT driver '{DriverId}': could not reconcile the Sparkplug primary-host claim for '{HostId}'; will retry.",
|
||||
_driverId,
|
||||
_hostId);
|
||||
if (_roleSource.ShouldActAsPrimary)
|
||||
{
|
||||
_claimFailures++;
|
||||
var delay = ClaimBackoffFor(_claimFailures);
|
||||
_nextClaimAttemptUtc = _timeProvider.GetUtcNow() + delay;
|
||||
|
||||
// One Warning per ATTEMPT, and attempts are now spaced by the backoff, so this is the
|
||||
// rate limit: 2 s, 4 s, 8 s … capped at MaxClaimBackoff rather than one line a second.
|
||||
_logger?.LogWarning(
|
||||
ex,
|
||||
"MQTT driver '{DriverId}': could not claim Sparkplug primary host '{HostId}' (attempt {Attempt}); retrying in {RetryIn}.",
|
||||
_driverId,
|
||||
_hostId,
|
||||
_claimFailures,
|
||||
delay);
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger?.LogWarning(
|
||||
ex,
|
||||
"MQTT driver '{DriverId}': could not reconcile the Sparkplug primary-host claim for '{HostId}'; will retry.",
|
||||
_driverId,
|
||||
_hostId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Capped exponential delay before claim attempt N+1: <c>2^N</c> seconds, at most <see cref="MaxClaimBackoff"/>.</summary>
|
||||
/// <param name="failures">Consecutive failed claim attempts so far (at least 1).</param>
|
||||
/// <returns>How long to wait before the next attempt.</returns>
|
||||
internal static TimeSpan ClaimBackoffFor(int failures)
|
||||
{
|
||||
var seconds = Math.Pow(2, Math.Min(failures, 16));
|
||||
return TimeSpan.FromSeconds(Math.Min(seconds, MaxClaimBackoff.TotalSeconds));
|
||||
}
|
||||
|
||||
private async Task ClaimAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (Volatile.Read(ref _connection) is not null)
|
||||
|
||||
@@ -238,6 +238,53 @@ public sealed class SparkplugPrimaryHostTests
|
||||
await host.DisposeAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// STAB-27 — a claim the broker keeps refusing used to be redialled (and Warning-logged) on every
|
||||
/// one-second role sample, forever. It now backs off. The clock is frozen, so the real one-second
|
||||
/// loop ticks several times without the backoff window ever opening; advancing the clock opens it.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_failing_claim_backs_off_instead_of_redialling_every_tick()
|
||||
{
|
||||
var clock = new FrozenClock();
|
||||
var host = new SparkplugPrimaryHost(
|
||||
Unreachable(), "drv-1", new FixedRole(shouldActAsPrimary: true), timeProvider: clock);
|
||||
|
||||
await host.StartAsync(TestContext.Current.CancellationToken);
|
||||
host.ClaimAttemptsForTests.ShouldBe(1);
|
||||
|
||||
// Several real loop ticks elapse while the (frozen) backoff window stays shut.
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(3200), TestContext.Current.CancellationToken);
|
||||
host.ClaimAttemptsForTests.ShouldBe(1, "no further claim while inside the backoff window");
|
||||
|
||||
// Positive control: once the window passes, the very next tick does try again.
|
||||
clock.Advance(TimeSpan.FromSeconds(5));
|
||||
(await WaitUntilAsync(() => host.ClaimAttemptsForTests >= 2, TimeSpan.FromSeconds(10)))
|
||||
.ShouldBeTrue("the claim must resume after the backoff window");
|
||||
|
||||
await host.DisposeAsync();
|
||||
}
|
||||
|
||||
/// <summary>The ladder is exponential from 2 s and capped, so a long outage settles at one attempt per cap.</summary>
|
||||
[Theory]
|
||||
[InlineData(1, 2)]
|
||||
[InlineData(2, 4)]
|
||||
[InlineData(3, 8)]
|
||||
[InlineData(4, 16)]
|
||||
[InlineData(5, 30)]
|
||||
[InlineData(40, 30)]
|
||||
public void Claim_backoff_is_exponential_and_capped(int failures, int expectedSeconds)
|
||||
=> SparkplugPrimaryHost.ClaimBackoffFor(failures).ShouldBe(TimeSpan.FromSeconds(expectedSeconds));
|
||||
|
||||
private sealed class FrozenClock : TimeProvider
|
||||
{
|
||||
private DateTimeOffset _now = new(2026, 10, 8, 0, 0, 0, TimeSpan.Zero);
|
||||
|
||||
public override DateTimeOffset GetUtcNow() => _now;
|
||||
|
||||
public void Advance(TimeSpan by) => _now += by;
|
||||
}
|
||||
|
||||
/// <summary>A broker that refuses instantly, so a connect attempt is observable but not slow.</summary>
|
||||
private static MqttDriverOptions Unreachable() =>
|
||||
Options() with { Host = "127.0.0.1", Port = 1, ConnectTimeoutSeconds = 1 };
|
||||
|
||||
Reference in New Issue
Block a user