fix(mqtt): back off the Sparkplug primary-host claim loop after failures (#602 STAB-27)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs

@@ -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(
+            if (_roleSource.ShouldActAsPrimary)
-                ex,
+            {
-                "MQTT driver '{DriverId}': could not reconcile the Sparkplug primary-host claim for '{HostId}'; will retry.",
+                _claimFailures++;
-                _driverId,
+                var delay = ClaimBackoffFor(_claimFailures);
-                _hostId);
+                _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)

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs

@@ -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 };