From 275b240ebdc9f9800159728e07e23ff7916931cb Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 07:29:08 -0400 Subject: [PATCH] fix(mqtt): back off the Sparkplug primary-host claim loop after failures (#602 STAB-27) Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- .../Sparkplug/SparkplugPrimaryHost.cs | 57 +++++++++++++++++-- .../SparkplugPrimaryHostTests.cs | 47 +++++++++++++++ 2 files changed, 99 insertions(+), 5 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 02c1d696..ba4a2c98 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 @@ -66,6 +66,14 @@ public sealed class SparkplugPrimaryHost : IAsyncDisposable /// How often the role gate is sampled. Bounds the failover republish latency. private static readonly TimeSpan RoleSampleInterval = TimeSpan.FromSeconds(1); + /// Ceiling on the delay between failed claim attempts (STAB-27). + internal static readonly TimeSpan MaxClaimBackoff = TimeSpan.FromSeconds(30); + + /// Consecutive failed claim attempts; reset by a successful claim or a stand-down. Reconcile-loop-owned. + 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); + } } } + /// Capped exponential delay before claim attempt N+1: 2^N seconds, at most . + /// Consecutive failed claim attempts so far (at least 1). + /// How long to wait before the next attempt. + 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) 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 0cffd5b5..cfdcdd5a 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 @@ -238,6 +238,53 @@ public sealed class SparkplugPrimaryHostTests await host.DisposeAsync(); } + /// + /// 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. + /// + [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(); + } + + /// The ladder is exponential from 2 s and capped, so a long outage settles at one attempt per cap. + [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; + } + /// A broker that refuses instantly, so a connect attempt is observable but not slow. private static MqttDriverOptions Unreachable() => Options() with { Host = "127.0.0.1", Port = 1, ConnectTimeoutSeconds = 1 };