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

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 07:29:08 -04:00
parent 1f4f4fca8f
commit 275b240ebd
2 changed files with 99 additions and 5 deletions
@@ -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)