v1.0.36

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

@@ -101,4 +101,33 @@ Task<BackgroundJobCancellationResult> CancelWaitingAsync(
     /// </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

@@ -78,6 +78,46 @@ public async Task<Guid> EnqueueAsync<TPayload>(
         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));
@@ -91,7 +131,10 @@ public async Task<Guid> EnqueueAsync<TPayload>(
         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);
@@ -155,19 +198,26 @@ public async Task<Guid> EnqueueAsync<TPayload>(
         // 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(
@@ -187,14 +237,26 @@ public async Task<Guid> EnqueueAsync<TPayload>(
             }
         }
 
-        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>
@@ -248,7 +310,10 @@ public async Task UpdateAsync(Guid id, string newSchedule, CancellationToken can
         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);
@@ -413,7 +478,10 @@ public async Task<bool> DeleteAsync(Guid id, CancellationToken cancellationToken
         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);

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 @@ public virtual async Task DispatchAsync(
         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);

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 @@ namespace BBT.Aether.BackgroundJob.Processing;
 /// <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,
@@ -24,6 +32,15 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
             "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 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 
             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;
 }