Background-job span gating, deferred arm handle, and jittered poll pacing

framework/src/BBT.Aether.Core/BBT/Aether/BackgroundJob/IBackgroundJobArmHandle.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace BBT.Aether.BackgroundJob;
+
+/// <summary>
+/// A deferred arm for an already-persisted background job: everything the external scheduler needs is
+/// captured in memory, so arming later costs exactly one scheduler call — no re-read of the job row and
+/// no extra status write.
+/// <para>
+/// Exists for callers that must persist the job inside a critical section but cannot afford to make the
+/// scheduler round-trip there. Holding a distributed lock across an external call makes that call the
+/// lock's hold time, serializing every other contender behind it.
+/// </para>
+/// <para>
+/// The row is already <c>Scheduled</c> when the handle is issued — optimistically, because the common
+/// case succeeds. <see cref="ArmAsync"/> reconciles a failure by rolling the row back to
+/// <c>Pending</c> so the arming poller reclaims it, which is the same contract the inline arm has.
+/// </para>
+/// </summary>
+public interface IBackgroundJobArmHandle
+{
+    /// <summary>The id of the persisted job this handle arms.</summary>
+    Guid JobId { get; }
+
+    /// <summary>
+    /// Arms the job in the external scheduler. Never throws: a failure is logged and the row is rolled
+    /// back to <c>Pending</c> for the arming poller. Safe to call once; calling it again re-schedules
+    /// the same job name, which the scheduler treats as an overwrite.
+    /// </summary>
+    Task ArmAsync(CancellationToken cancellationToken = default);
+}

framework/src/BBT.Aether.Core/BBT/Aether/BackgroundJob/IBackgroundJobService.cs

@@ -54,17 +54,17 @@
     /// </returns>
     /// <exception cref="ArgumentNullException">Thrown when any required parameter is null.</exception>
     /// <exception cref="InvalidOperationException">Thrown when the job scheduling service is unavailable.</exception>
     Task<Guid> EnqueueAsync<TPayload>(
         string handlerName,
         string jobName,
         TPayload payload,
         string schedule,
         Dictionary<string, object>? metadata = null,
         JobScheduleFailurePolicy? failurePolicyOptions = null,
         bool directly = false,
         Guid? jobId = null,
         BBT.Aether.Domain.Entities.JobKind? kind = null,
         CancellationToken cancellationToken = default);
 
     /// <summary>
     /// Atomically updates the schedule of a Pending, Scheduled, or Retrying background job.
@@ -101,4 +101,33 @@
     /// </returns>
     /// <exception cref="ArgumentException">Thrown when id is empty.</exception>
     Task<bool> DeleteAsync(Guid id, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Persists the job and returns a handle that arms it later, instead of arming as part of this call.
+    /// <para>
+    /// For callers that write the job inside a critical section (a distributed lock, a short
+    /// transaction) but must keep the scheduler round-trip out of it. Persisting is cheap and local;
+    /// the scheduler call is neither, and inside a lock it becomes the lock's hold time.
+    /// </para>
+    /// <para>
+    /// Same parameters and same persistence semantics as <see cref="EnqueueAsync{TPayload}"/> with
+    /// <c>directly: true</c> — the row lands <c>Scheduled</c> and an arm failure rolls it back to
+    /// <c>Pending</c> for the arming poller. The only difference is WHEN the scheduler is called, which
+    /// the caller now decides by invoking <see cref="IBackgroundJobArmHandle.ArmAsync"/>.
+    /// </para>
+    /// <para>
+    /// Call <c>ArmAsync</c> only after the work that justified the critical section has committed. The
+    /// handle carries the payload in memory, so arming costs one scheduler call and no database access.
+    /// </para>
+    /// </summary>
+    Task<IBackgroundJobArmHandle> EnqueueWithDeferredArmAsync<TPayload>(
+        string handlerName,
+        string jobName,
+        TPayload payload,
+        string schedule,
+        Dictionary<string, object>? metadata = null,
+        JobScheduleFailurePolicy? failurePolicyOptions = null,
+        Guid? jobId = null,
+        BBT.Aether.Domain.Entities.JobKind? kind = null,
+        CancellationToken cancellationToken = default);
 }

framework/src/BBT.Aether.Infrastructure/BBT/Aether/BackgroundJob/BackgroundJobService.cs

@@ -23,16 +23,16 @@
 /// Uses UoW pattern for transactional consistency.
 /// Wraps job payloads in CloudEventEnvelope to carry schema context and metadata.
 /// </summary>
 public sealed class BackgroundJobService(
     IJobStore jobStore,
     IJobScheduler jobScheduler,
     IUnitOfWorkManager uowManager,
     IGuidGenerator guidGenerator,
     IClock clock,
     ICurrentSchema currentSchema,
     IEventSerializer eventSerializer,
     BackgroundJobOptions options,
     ILogger<BackgroundJobService> logger)
     : IBackgroundJobService
 {
     private const string Source = "urn:background-job";
@@ -78,6 +78,46 @@
         Guid? jobId = null,
         JobKind? kind = null,
         CancellationToken cancellationToken = default)
+    {
+        var (enqueuedId, _) = await EnqueueCoreAsync(
+            handlerName, jobName, payload, schedule, metadata, failurePolicyOptions,
+            directly, jobId, kind, deferArm: false, cancellationToken);
+        return enqueuedId;
+    }
+
+    /// <inheritdoc/>
+    public async Task<IBackgroundJobArmHandle> EnqueueWithDeferredArmAsync<TPayload>(
+        string handlerName,
+        string jobName,
+        TPayload payload,
+        string schedule,
+        Dictionary<string, object>? metadata = null,
+        JobScheduleFailurePolicy? failurePolicyOptions = null,
+        Guid? jobId = null,
+        JobKind? kind = null,
+        CancellationToken cancellationToken = default)
+    {
+        // directly: true so the row lands Scheduled, exactly as the inline path leaves it — no later
+        // status write is needed. The one difference is that the scheduler is not called here; the
+        // returned handle calls it, carrying the payload in memory so no re-read is needed either.
+        var (deferredId, arm) = await EnqueueCoreAsync(
+            handlerName, jobName, payload, schedule, metadata, failurePolicyOptions,
+            directly: true, jobId, kind, deferArm: true, cancellationToken);
+        return new DeferredArmHandle(deferredId, arm!);
+    }
+
+    private async Task<(Guid JobId, Func<CancellationToken, Task>? Arm)> EnqueueCoreAsync<TPayload>(
+        string handlerName,
+        string jobName,
+        TPayload payload,
+        string schedule,
+        Dictionary<string, object>? metadata,
+        JobScheduleFailurePolicy? failurePolicyOptions,
+        bool directly,
+        Guid? jobId,
+        JobKind? kind,
+        bool deferArm,
+        CancellationToken cancellationToken)
     {
         if (string.IsNullOrWhiteSpace(handlerName))
             throw new ArgumentNullException(nameof(handlerName));
@@ -85,13 +125,16 @@
         if (string.IsNullOrWhiteSpace(jobName))
             throw new ArgumentNullException(nameof(jobName));
 
         if (payload == null)
             throw new ArgumentNullException(nameof(payload));
 
         if (string.IsNullOrWhiteSpace(schedule))
             throw new ArgumentNullException(nameof(schedule));
 
-        using var activity = InfrastructureActivitySource.Source.StartActivity(
+        // Diagnostic, not business: enqueueing a job is plumbing around the work, and in the
+        // default Business profile it only adds depth to an already deep trace. The job's own
+        // execution span (BackgroundJob.Execute) is the one that carries meaning and stays.
+        using var activity = InfrastructureActivitySource.StartDiagnosticActivity(
             "BackgroundJob.Enqueue",
             ActivityKind.Producer,
             Activity.Current?.Context ?? default);
@@ -100,9 +143,9 @@
         activity?.SetTag("job.name", jobName);
         activity?.SetTag("job.schedule", schedule);
 
         logger.LogInformation(
             "Enqueueing job handler '{HandlerName}' with job name '{JobName}' and schedule '{Schedule}'",
             handlerName, jobName, schedule);
 
         // Create job entity. A caller-supplied id (when present) lets the caller reuse a single
         // correlation id for its own tracking row; otherwise generate one.
@@ -147,7 +190,7 @@
             ExtraProperties = extraProperties
         };
 
         // TODO(jobs): thread failurePolicyOptions through the arming poller. The poller currently arms
         // with a default failure policy; the caller-supplied policy is not yet persisted/honored. Kept in
         // the signature so callers don't break and so a later task can wire it through ExtraProperties.
         // The `directly` path below DOES honor failurePolicyOptions, via ScheduleAsync.
@@ -155,30 +198,37 @@
         // Bytes for the scheduler (the `directly` arm path). Equivalent to the JSON the poller arms with.
         var payloadBytes = eventSerializer.Serialize(envelope);
 
+        // Deferred arm: hand the caller a closure over everything the scheduler needs. Same call the
+        // inline path makes, same failure handling — only the timing is the caller's to choose.
+        Func<CancellationToken, Task>? armAction = deferArm
+            ? ct => ArmNowAsync(handlerName, jobName, schedule, payloadBytes,
+                failurePolicyOptions, effectiveJobId, ct)
+            : null;
+
         // Atomic-ambient: when the caller has an ambient UoW, persist into it (commits with their business
         // transaction — a rollback discards the row). Otherwise open a short own transaction.
         if (uowManager.Current is { } ambient)
         {
             await jobStore.SaveAsync(jobInfo, cancellationToken);
-            if (directly)
+            if (directly && !deferArm)
                 ambient.OnCompleted(_ => ArmNowAsync(handlerName, jobName, schedule, payloadBytes,
                     failurePolicyOptions, effectiveJobId, CancellationToken.None));
             logger.LogInformation(
                 "Enqueued {Status} job '{HandlerName}'/'{JobName}' into ambient UoW. Id: {Id}",
                 jobInfo.Status, handlerName, jobName, effectiveJobId);
             activity?.SetStatus(ActivityStatusCode.Ok);
-            return effectiveJobId;
+            return (effectiveJobId, armAction);
         }
 
         await using (var uow = uowManager.Begin(
                          new UnitOfWorkOptions { Scope = UnitOfWorkScopeOption.RequiresNew, IsTransactional = true }))
         {
             try
             {
                 await jobStore.SaveAsync(jobInfo, cancellationToken);
                 await uow.CommitAsync(cancellationToken);
             }
             catch (Exception ex)
             {
                 logger.LogError(ex, "Failed to enqueue job '{HandlerName}'/'{JobName}'", handlerName, jobName);
                 RecordException(activity, ex);
@@ -187,14 +237,26 @@
             }
         }
 
-        if (directly)
+        if (directly && !deferArm)
             await ArmNowAsync(handlerName, jobName, schedule, payloadBytes, failurePolicyOptions,
                 effectiveJobId, cancellationToken);
         logger.LogInformation(
             "Enqueued {Status} job '{HandlerName}'/'{JobName}'. Id: {Id}",
             jobInfo.Status, handlerName, jobName, effectiveJobId);
         activity?.SetStatus(ActivityStatusCode.Ok);
-        return effectiveJobId;
+        return (effectiveJobId, armAction);
+    }
+
+    /// <summary>
+    /// Closure-backed <see cref="IBackgroundJobArmHandle"/>. Holds the scheduler arguments captured at
+    /// enqueue time, so arming needs neither a job-row read nor a status write.
+    /// </summary>
+    private sealed class DeferredArmHandle(Guid jobId, Func<CancellationToken, Task> arm)
+        : IBackgroundJobArmHandle
+    {
+        public Guid JobId { get; } = jobId;
+
+        public Task ArmAsync(CancellationToken cancellationToken = default) => arm(cancellationToken);
     }
 
     /// <summary>
@@ -216,8 +278,8 @@
             IUnitOfWork? rollbackUow = null;
             try
             {
                 rollbackUow = uowManager.Begin(
                     new UnitOfWorkOptions { Scope = UnitOfWorkScopeOption.RequiresNew, IsTransactional = true });
                 await jobStore.TryTransitionStatusAsync(jobId, BackgroundJobStatus.Scheduled,
                     BackgroundJobStatus.Pending, ct);
                 await rollbackUow.CommitAsync(ct);
@@ -248,7 +310,10 @@
         if (string.IsNullOrWhiteSpace(newSchedule))
             throw new ArgumentNullException(nameof(newSchedule));
 
-        using var activity = InfrastructureActivitySource.Source.StartActivity(
+        // Diagnostic, not business: enqueueing a job is plumbing around the work, and in the
+        // default Business profile it only adds depth to an already deep trace. The job's own
+        // execution span (BackgroundJob.Execute) is the one that carries meaning and stays.
+        using var activity = InfrastructureActivitySource.StartDiagnosticActivity(
             "BackgroundJob.Update",
             ActivityKind.Producer,
             Activity.Current?.Context ?? default);
@@ -256,7 +321,7 @@
         activity?.SetTag("job.id", id.ToString());
         activity?.SetTag("job.schedule", newSchedule);
 
         logger.LogInformation("Updating job with entity id '{Id}' to new schedule '{NewSchedule}'", id, newSchedule);
 
         if (jobStore is not IJobRescheduleStore rescheduleStore)
         {
@@ -274,13 +339,13 @@
             var result = await rescheduleStore.TryRescheduleWaitingAsync(
                 id, newSchedule, kind, nextRetryAt, cancellationToken);
             EnsureRescheduled(id, result, activity);
             logger.LogInformation("Successfully updated job with entity id '{Id}'", id);
             activity?.SetStatus(ActivityStatusCode.Ok);
             return;
         }
 
         await using var uow = uowManager.Begin(
             new UnitOfWorkOptions { Scope = UnitOfWorkScopeOption.RequiresNew, IsTransactional = true });
         try
         {
             var result = await rescheduleStore.TryRescheduleWaitingAsync(
@@ -290,10 +355,10 @@
             // Commit transaction
             await uow.CommitAsync(cancellationToken);
 
             logger.LogInformation("Successfully updated job with entity id '{Id}'", id);
             activity?.SetStatus(ActivityStatusCode.Ok);
         }
         catch (Exception ex)
         {
             logger.LogError(ex, "Failed to update job with entity id '{Id}'", id);
             RecordException(activity, ex);
@@ -345,11 +410,11 @@
 
         BackgroundJobCancellationSnapshot? snapshot;
         BackgroundJobCancellationResult result;
         await using (var uow = uowManager.Begin(new UnitOfWorkOptions
                      {
                          Scope = UnitOfWorkScopeOption.RequiresNew,
                          IsTransactional = true
                      }))
         {
             snapshot = await jobStore.GetCancellationSnapshotAsync(id, cancellationToken);
             result = snapshot is null
@@ -413,14 +478,17 @@
         if (id == Guid.Empty)
             throw new ArgumentException("Id cannot be empty.", nameof(id));
 
-        using var activity = InfrastructureActivitySource.Source.StartActivity(
+        // Diagnostic, not business: enqueueing a job is plumbing around the work, and in the
+        // default Business profile it only adds depth to an already deep trace. The job's own
+        // execution span (BackgroundJob.Execute) is the one that carries meaning and stays.
+        using var activity = InfrastructureActivitySource.StartDiagnosticActivity(
             "BackgroundJob.Delete",
             ActivityKind.Producer,
             Activity.Current?.Context ?? default);
 
         activity?.SetTag("job.id", id.ToString());
 
         logger.LogInformation("Deleting job with entity id '{Id}'", id);
 
         if (uowManager.Current is { })
         {
@@ -437,13 +505,13 @@
             await jobScheduler.DeleteAsync(jobInfo.HandlerName, jobInfo.JobName, cancellationToken);
             await jobStore.UpdateStatusAsync(id, BackgroundJobStatus.Cancelled, clock.UtcNow,
                 cancellationToken: cancellationToken);
             logger.LogInformation("Successfully deleted job with entity id '{Id}'", id);
             activity?.SetStatus(ActivityStatusCode.Ok);
             return true;
         }
 
         await using var uow = uowManager.Begin(
             new UnitOfWorkOptions { Scope = UnitOfWorkScopeOption.RequiresNew, IsTransactional = true });
         try
         {
             // Load job from store
@@ -468,11 +536,11 @@
             // Commit transaction
             await uow.CommitAsync(cancellationToken);
 
             logger.LogInformation("Successfully deleted job with entity id '{Id}'", id);
             activity?.SetStatus(ActivityStatusCode.Ok);
             return true;
         }
         catch (Exception ex)
         {
             logger.LogError(ex, "Failed to delete job with entity id '{Id}'", id);
             RecordException(activity, ex);

framework/src/BBT.Aether.Infrastructure/BBT/Aether/BackgroundJob/Dapr/DaprJobScheduler.cs

@@ -146,7 +146,9 @@ await daprJobsClient.DeleteJobAsync(
 
     private static Activity? StartSchedulerActivity(string operationName, string handlerName, string jobName)
     {
-        var activity = InfrastructureActivitySource.Source.StartActivity(
+        // Diagnostic: the scheduler round-trip is infrastructure detail. Business traces care that
+        // the job ran (BackgroundJob.Execute), not about the Schedule/Delete calls that armed it.
+        var activity = InfrastructureActivitySource.StartDiagnosticActivity(
             operationName,
             ActivityKind.Client,
             Activity.Current?.Context ?? default);

framework/src/BBT.Aether.Infrastructure/BBT/Aether/BackgroundJob/JobDispatcher.cs

@@ -32,7 +32,8 @@
         if (string.IsNullOrWhiteSpace(jobName))
             throw new ArgumentNullException(nameof(jobName));
 
-        using var activity = InfrastructureActivitySource.Source.StartActivity(
+        // Diagnostic: pure dispatch plumbing between BackgroundJob.Execute and the handler.
+        using var activity = InfrastructureActivitySource.StartDiagnosticActivity(
             "BackgroundJob.Dispatch",
             ActivityKind.Internal,
             Activity.Current?.Context ?? default);
@@ -87,8 +88,8 @@
             var recorded = await RecordSuccessAsync(scope, c, jobName, activity, cancellationToken);
             if (recorded)
             {
                 logger.LogInformation("Successfully completed handler '{HandlerName}' for job id '{JobId}'",
                     c.HandlerName, c.JobId);
             }
         }
         catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
@@ -97,11 +98,11 @@
             var jobStore = scope.ServiceProvider.GetRequiredService<IJobStore>();
             var jobScheduler = scope.ServiceProvider.GetRequiredService<IJobScheduler>();
 
             logger.LogWarning("Handler '{HandlerName}' for job id '{JobId}' was cancelled", c.HandlerName, c.JobId);
             activity?.SetTag("job.status", "cancelled");
             activity?.SetStatus(ActivityStatusCode.Ok);
             await using (var cancelUow = uowManager.Begin(
                 new UnitOfWorkOptions { Scope = UnitOfWorkScopeOption.RequiresNew, IsTransactional = true }))
             {
                 await jobStore.TryRecordTerminalAsync(c.JobId, c.Token, BackgroundJobStatus.Cancelled,
                     clock.UtcNow, "Job was cancelled", cancellationToken);
@@ -142,8 +143,8 @@
         var jobStore = scope.ServiceProvider.GetRequiredService<IJobStore>();
         var jobScheduler = scope.ServiceProvider.GetRequiredService<IJobScheduler>();
 
         await using var claimUow = uowManager.Begin(
             new UnitOfWorkOptions { Scope = UnitOfWorkScopeOption.RequiresNew, IsTransactional = true });
 
         var jobInfo = await jobStore.GetByJobNameAsync(jobName, cancellationToken);
         if (jobInfo is null)
@@ -176,8 +177,8 @@
         await claimUow.CommitAsync(cancellationToken);
         if (!claimed)
         {
             logger.LogInformation(
                 "Job id '{JobId}' was not Scheduled (already claimed or late delivery); skipping", jobInfo.Id);
             activity?.SetTag("job.status", "skipped");
             activity?.SetStatus(ActivityStatusCode.Ok);
             return null;
@@ -202,8 +203,8 @@
         var jobScheduler = scope.ServiceProvider.GetRequiredService<IJobScheduler>();
 
         bool recorded;
         await using (var doneUow = uowManager.Begin(
             new UnitOfWorkOptions { Scope = UnitOfWorkScopeOption.RequiresNew, IsTransactional = true }))
         {
             recorded = claim.Kind == JobKind.Recurring
                 ? await jobStore.TryReturnToScheduledAsync(claim.JobId, claim.Token, clock.UtcNow, null, cancellationToken)
@@ -246,8 +247,8 @@
         var willRetry = claim.Kind == JobKind.OneShot && claim.RetryCount + 1 <= claim.MaxRetryCount;
 
         bool recorded;
         await using (var failUow = uowManager.Begin(
             new UnitOfWorkOptions { Scope = UnitOfWorkScopeOption.RequiresNew, IsTransactional = true }))
         {
             if (claim.Kind == JobKind.Recurring)
             {

framework/src/BBT.Aether.Infrastructure/BBT/Aether/BackgroundJob/Processing/BackgroundJobArmingHostedService.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using BBT.Aether.Polling;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 
@@ -9,8 +10,15 @@
 /// <summary>
 /// Hosted service that drives <see cref="BackgroundJobArmingProcessor"/> on a timer. Each tick runs one
 /// arming pass; exceptions per tick are caught and logged so a transient failure never tears down the
-/// loop. The delay between ticks is <see cref="BackgroundJobOptions.ArmingInterval"/>. Registered by the
-/// DI wiring (see AddAetherBackgroundJob); not auto-registered here.
+/// loop. The delay between ticks is <see cref="BackgroundJobOptions.ArmingInterval"/>, jittered.
+/// Registered by the DI wiring (see AddAetherBackgroundJob); not auto-registered here.
+/// <para>
+/// The interval is fixed — there is no adaptive backoff here, because an unarmed job must be picked up
+/// within a bounded time regardless of how quiet the system is. That makes jitter the only thing
+/// keeping replicas apart: without it, pods started together by a rolling deployment run every pass in
+/// lockstep, turning each tick into a burst of simultaneous claim queries over the same rows. A random
+/// startup offset spreads the first pass as well.
+/// </para>
 /// </summary>
 public class BackgroundJobArmingHostedService(
     BackgroundJobArmingProcessor processor,
@@ -20,10 +28,19 @@
     /// <inheritdoc />
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         logger.LogInformation(
             "Background-job arming poller started (interval {Interval}, schema {Schema}).",
             options.ArmingInterval, options.Schema);
 
+        try
+        {
+            await Task.Delay(PollingDelay.StartupOffset(options.ArmingInterval), stoppingToken);
+        }
+        catch (OperationCanceledException)
+        {
+            return;
+        }
+
         while (!stoppingToken.IsCancellationRequested)
         {
             try
@@ -41,7 +58,7 @@
 
             try
             {
-                await Task.Delay(options.ArmingInterval, stoppingToken);
+                await Task.Delay(PollingDelay.Jitter(options.ArmingInterval), stoppingToken);
             }
             catch (OperationCanceledException)
             {

framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/InboxBackgroundService.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using BBT.Aether.Polling;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 
@@ -13,6 +14,18 @@ public sealed class InboxBackgroundService(
 {
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
+        // Spread the first poll of replicas that booted together, so a rolling deployment does not
+        // leave the whole fleet polling on the same tick.
+        try
+        {
+            await Task.Delay(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppingToken)
+                .ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+        {
+            return;
+        }
+
         var delay = options.IdlePollingInterval;
 
         while (!stoppingToken.IsCancellationRequested)
@@ -21,8 +34,8 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             {
                 var processed = await processor.RunAsync(stoppingToken);
                 delay = processed > 0
-                    ? options.BusyPollingInterval
-                    : Min(delay * 2, options.MaxPollingInterval);
+                    ? PollingDelay.OnProcessed(options.BusyPollingInterval)
+                    : PollingDelay.OnEmpty(delay, options.MaxPollingInterval);
             }
             catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
             {
@@ -31,12 +44,12 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             catch (Exception ex)
             {
                 logger.LogError(ex, "Inbox background service error");
-                delay = options.MaxPollingInterval;
+                // One step back, not straight to the cap: a transient fault must not stall every
+                // replica for a full maximum interval.
+                delay = PollingDelay.OnError(delay, options.IdlePollingInterval, options.MaxPollingInterval);
             }
 
-            await Task.Delay(delay, stoppingToken).ConfigureAwait(false);
+            await Task.Delay(PollingDelay.Jitter(delay), stoppingToken).ConfigureAwait(false);
         }
     }
-
-    private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b;
 }

framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxBackgroundService.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using BBT.Aether.Polling;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 
@@ -13,6 +14,18 @@ public sealed class OutboxBackgroundService(
 {
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
+        // Spread the first poll of replicas that booted together, so a rolling deployment does not
+        // leave the whole fleet polling on the same tick.
+        try
+        {
+            await Task.Delay(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppingToken)
+                .ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+        {
+            return;
+        }
+
         var delay = options.IdlePollingInterval;
 
         while (!stoppingToken.IsCancellationRequested)
@@ -21,8 +34,8 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             {
                 var processed = await processor.RunAsync(stoppingToken);
                 delay = processed > 0
-                    ? options.BusyPollingInterval
-                    : Min(delay * 2, options.MaxPollingInterval);
+                    ? PollingDelay.OnProcessed(options.BusyPollingInterval)
+                    : PollingDelay.OnEmpty(delay, options.MaxPollingInterval);
             }
             catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
             {
@@ -31,12 +44,12 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             catch (Exception ex)
             {
                 logger.LogError(ex, "Outbox background service error");
-                delay = options.MaxPollingInterval;
+                // One step back, not straight to the cap: a transient fault must not stall every
+                // replica for a full maximum interval.
+                delay = PollingDelay.OnError(delay, options.IdlePollingInterval, options.MaxPollingInterval);
             }
 
-            await Task.Delay(delay, stoppingToken).ConfigureAwait(false);
+            await Task.Delay(PollingDelay.Jitter(delay), stoppingToken).ConfigureAwait(false);
         }
     }
-
-    private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b;
 }