Keep job leases through transient errors, stop handlers that lost them, and release jobs on shutdown

CONTRIBUTING.md

@@ -108,7 +108,7 @@ Work that outlives a request, such as a bulk operation or a check run, is a job
 
 A handler implements `IJobHandler` (an extension point) with the type name its jobs store, which never changes once released. It reads the payload from `JobContext.Job`, resolves services from `JobContext.Services`, a scope of its own, and reports progress with `ReportProgressAsync`, which also keeps its lease; when that returns false another worker has the job, and the handler stops. A returned failure ends the job as `failed`. An exception retries it after `JobQueue__RetryDelay`, doubled for each attempt up to `JobQueue__MaxRetryDelay`, and after its last attempt the job is `dead`. Work in a handler therefore has to be safe to run again.
 
-`JobQueue__WorkerCount` workers in the host claim due jobs every `JobQueue__PollingInterval`. A claim holds a job for `JobQueue__LeaseDuration`, which the worker renews while the handler runs, and a job whose worker stopped is claimed again once the lease runs out, counting another attempt. Claims are conditional updates on the row, so with any number of workers, in one host or several, each job runs once at a time.
+`JobQueue__WorkerCount` workers in the host claim due jobs every `JobQueue__PollingInterval`. A claim holds a job for `JobQueue__LeaseDuration`, which the worker renews at a third of it while the handler runs. A renewal that fails is tried again while the lease holds, and no wait or renewal lasts past the lease's end. The handler's cancellation token fires as soon as the lease is lost or its end passes with no renewal, because another worker may then claim the job; a handler should honour it. A job whose worker stopped is claimed again once the lease runs out, counting another attempt. A host that stops while a handler runs puts the job back in the queue with its attempt given back. Claims are conditional updates on the row, so with any number of workers, in one host or several, each job runs once at a time.
 
 ## Model access
 

src/FluentCMS.Core/Jobs/IJobLeases.cs

@@ -22,4 +22,7 @@ public interface IJobLeases
 
     // Queued again after the delay while attempts remain, dead otherwise.
     Task<bool> RetryAsync(Guid jobId, string owner, string error, TimeSpan delay, CancellationToken cancellationToken = default);
+
+    // Queued again at once with the attempt given back, for a job its worker stopped on purpose, such as when the host shuts down.
+    Task<bool> ReleaseAsync(Guid jobId, string owner, CancellationToken cancellationToken = default);
 }

src/FluentCMS.Host/Jobs/JobQueueWorker.cs

@@ -8,6 +8,7 @@ namespace FluentCMS.Host.Jobs;
 public sealed class JobQueueWorker(
     IServiceScopeFactory scopes,
     IOptions<JobQueueOptions> options,
+    TimeProvider clock,
     ILogger<JobQueueWorker> logger) : BackgroundService
 {
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -55,6 +56,8 @@ private async Task<bool> RunNextAsync(string owner, JobQueueOptions settings, Ca
         await using var leaseScope = scopes.CreateAsyncScope();
         var leases = leaseScope.ServiceProvider.GetRequiredService<IJobLeases>();
 
+        // Taken before the claim writes the lease, so the heartbeat never counts on more of it than the claim gave.
+        var claimedAt = clock.GetUtcNow();
         var job = await leases.ClaimAsync(owner, settings.LeaseDuration, stoppingToken);
         if (job is null)
         {
@@ -78,14 +81,16 @@ private async Task<bool> RunNextAsync(string owner, JobQueueOptions settings, Ca
             runScope.ServiceProvider,
             (progress, cancellationToken) => leases.ReportProgressAsync(job.Id, owner, progress, settings.LeaseDuration, cancellationToken));
 
-        // The lease is kept while the handler works, so a long step without progress does not hand the job to another worker.
+        // The lease is kept while the handler works, so a long step without progress does not hand the job to another worker. Once it
+        // is lost, the handler is stopped through its token, since another worker may already be running the job.
+        using var running = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
         using var heartbeat = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
-        var keeping = KeepLeaseAsync(job.Id, owner, settings.LeaseDuration, heartbeat.Token);
+        var keeping = KeepAsync(job, owner, claimedAt + settings.LeaseDuration, settings.LeaseDuration, running, heartbeat.Token);
 
         // Once the handler is done, how it ended is recorded even if the host is stopping, so finished work is not run again.
         try
         {
-            var result = await handler.RunAsync(context, stoppingToken);
+            var result = await handler.RunAsync(context, running.Token);
             await heartbeat.CancelAsync();
             await keeping;
 
@@ -100,9 +105,22 @@ private async Task<bool> RunNextAsync(string owner, JobQueueOptions settings, Ca
         }
         catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
         {
-            // The host is stopping mid-job: the lease runs out and another worker runs the job again.
+            await heartbeat.CancelAsync();
+            await keeping;
+
+            // The host is stopping mid-job: the job goes back to the queue with its attempt given back, so stopping never uses one up.
+            if (!await ReleaseAsync(leases, job, owner))
+            {
+                logger.LogWarning("Job {JobId} could not be put back in the queue as the host stopped; it runs again once its lease runs out.", job.Id);
+            }
+
             throw;
         }
+        catch (OperationCanceledException) when (running.IsCancellationRequested)
+        {
+            // The lease was lost and the handler stopped: the job is the next claimant's, so nothing is recorded for this attempt.
+            await keeping;
+        }
         catch (Exception exception)
         {
             await heartbeat.CancelAsync();
@@ -111,38 +129,111 @@ private async Task<bool> RunNextAsync(string owner, JobQueueOptions settings, Ca
             logger.LogWarning(exception, "Job {JobId} of the type {JobType} failed on attempt {Attempt} of {MaxAttempts}.", job.Id, job.Type, job.Attempts, job.MaxAttempts);
             await leases.RetryAsync(job.Id, owner, exception.Message, RetryDelay(settings, job.Attempts), CancellationToken.None);
         }
+
         return true;
     }
 
-    // Renews the lease at a third of its duration, each time in a scope of its own, until the handler is done.
-    private async Task KeepLeaseAsync(Guid jobId, string owner, TimeSpan duration, CancellationToken cancellationToken)
+    // Each renewal has a scope of its own; a lost lease stops the handler.
+    private async Task KeepAsync(Job job, string owner, DateTimeOffset heldUntil, TimeSpan duration, CancellationTokenSource running, CancellationToken cancellationToken)
     {
-        try
-        {
-            while (true)
+        var kept = await KeepLeaseAsync(
+            async renewing =>
             {
-                await Task.Delay(duration / 3, cancellationToken);
                 await using var scope = scopes.CreateAsyncScope();
-                if (!await scope.ServiceProvider.GetRequiredService<IJobLeases>().RenewAsync(jobId, owner, duration, cancellationToken))
+                return await scope.ServiceProvider.GetRequiredService<IJobLeases>().RenewAsync(job.Id, owner, duration, renewing);
+            },
+            heldUntil,
+            duration,
+            clock,
+            logger,
+            cancellationToken);
+
+        if (!kept)
+        {
+            logger.LogWarning("Job {JobId} of the type {JobType} lost its lease on attempt {Attempt}; its handler is stopped.", job.Id, job.Type, job.Attempts);
+            await running.CancelAsync();
+        }
+    }
+
+    // Renews the lease at a third of its duration until the token says the handler is done, and is then true. A renewal that throws is
+    // tried again while the lease holds, so one transient error does not give the job away. False once a renewal is refused, because
+    // another worker holds the lease, or as soon as the lease's deadline passes without a renewal: no wait or renewal outlasts it.
+    public static async Task<bool> KeepLeaseAsync(
+        Func<CancellationToken, Task<bool>> renew,
+        DateTimeOffset heldUntil,
+        TimeSpan duration,
+        TimeProvider clock,
+        ILogger logger,
+        CancellationToken cancellationToken)
+    {
+        while (true)
+        {
+            // A beat, or what is left of the lease when that is less, so a failed renewal is never retried past the deadline.
+            var left = heldUntil - clock.GetUtcNow();
+            var wait = left < duration / 3 ? left : duration / 3;
+            try
+            {
+                await Task.Delay(wait > TimeSpan.Zero ? wait : TimeSpan.Zero, clock, cancellationToken);
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                return true;
+            }
+
+            var renewingAt = clock.GetUtcNow();
+            if (renewingAt >= heldUntil)
+            {
+                logger.LogError("A job's lease ran out at {HeldUntil} with no renewal.", heldUntil);
+                return false;
+            }
+
+            using var attempt = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            attempt.CancelAfter(heldUntil - renewingAt);
+            try
+            {
+                if (!await renew(attempt.Token))
                 {
-                    return;
+                    return false;
                 }
+
+                heldUntil = renewingAt + duration;
+            }
+            catch (Exception) when (cancellationToken.IsCancellationRequested)
+            {
+                // The handler is done, so how the renewal ended no longer matters.
+                return true;
+            }
+            catch (Exception exception)
+            {
+                logger.LogWarning(exception, "A job's lease could not be renewed; it holds until {HeldUntil} and is tried again.", heldUntil);
             }
         }
-        catch (OperationCanceledException)
+    }
+
+    // The update is guarded by the owner, so a worker that lost the lease meanwhile changes nothing.
+    private async Task<bool> ReleaseAsync(IJobLeases leases, Job job, string owner)
+    {
+        try
         {
-            // The handler is done.
+            return await leases.ReleaseAsync(job.Id, owner, CancellationToken.None);
         }
         catch (Exception exception)
         {
-            logger.LogWarning(exception, "The lease on job {JobId} could not be renewed.", jobId);
+            logger.LogWarning(exception, "Job {JobId} could not be released.", job.Id);
+            return false;
         }
     }
 
-    // Doubled for each attempt after the first, so a failing dependency gets room to recover.
+    // Doubled for each attempt after the first, so a failing dependency gets room to recover. Doubling stops at the ceiling, so no
+    // attempt count overflows.
     public static TimeSpan RetryDelay(JobQueueOptions settings, int attempt)
     {
-        var delay = settings.RetryDelay * Math.Pow(2, Math.Max(0, attempt - 1));
+        var delay = settings.RetryDelay;
+        for (var doubled = 1; doubled < attempt && delay > TimeSpan.Zero && delay < settings.MaxRetryDelay; doubled++)
+        {
+            delay = delay >= settings.MaxRetryDelay / 2 ? settings.MaxRetryDelay : delay + delay;
+        }
+
         return delay < settings.MaxRetryDelay ? delay : settings.MaxRetryDelay;
     }
 

src/FluentCMS.Infrastructure/Jobs/JobLeases.cs

@@ -135,6 +135,22 @@ public async Task<bool> RetryAsync(Guid jobId, string owner, string error, TimeS
             cancellationToken);
     }
 
+    public Task<bool> ReleaseAsync(Guid jobId, string owner, CancellationToken cancellationToken = default)
+    {
+        var now = clock.GetUtcNow();
+
+        return UpdateHeldAsync(
+            jobId,
+            owner,
+            setters => setters
+                .SetProperty(row => row.State, JobState.Queued)
+                .SetProperty(row => row.Attempts, row => row.Attempts - 1)
+                .SetProperty(row => row.ScheduledAt, now)
+                .SetProperty(row => row.LeaseOwner, (string?)null)
+                .SetProperty(row => row.LeaseUntil, (DateTimeOffset?)null),
+            cancellationToken);
+    }
+
     private Task<bool> FinishAsync(Guid jobId, string owner, JobState state, string? error, CancellationToken cancellationToken)
     {
         var now = clock.GetUtcNow();