From a9a6b54193618d75cab5dace786f2debf672b875 Mon Sep 17 00:00:00 2001 From: Amir Pournasserian Date: Mon, 19 Oct 2026 02:36:03 -0400 Subject: [PATCH 1/3] Keep job leases through transient errors, stop handlers that lost them, and release jobs on shutdown --- CONTRIBUTING.md | 2 +- src/FluentCMS.Core/Jobs/IJobLeases.cs | 3 + src/FluentCMS.Host/Jobs/JobQueueWorker.cs | 105 ++++++- .../Jobs/JobLeases.cs | 16 ++ .../Jobs/JobQueueWorkerTests.cs | 266 +++++++++++++++++- 5 files changed, 366 insertions(+), 26 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 6e6cb5805..cd07cbde8 100644 --- a/CONTRIBUTING.md +++ b/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. Once the lease is lost, or runs out with no renewal, the handler's cancellation token fires, because another worker may 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 diff --git a/src/FluentCMS.Core/Jobs/IJobLeases.cs b/src/FluentCMS.Core/Jobs/IJobLeases.cs index 438802c04..03c3525f0 100644 --- a/src/FluentCMS.Core/Jobs/IJobLeases.cs +++ b/src/FluentCMS.Core/Jobs/IJobLeases.cs @@ -22,4 +22,7 @@ public interface IJobLeases // Queued again after the delay while attempts remain, dead otherwise. Task 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 ReleaseAsync(Guid jobId, string owner, CancellationToken cancellationToken = default); } diff --git a/src/FluentCMS.Host/Jobs/JobQueueWorker.cs b/src/FluentCMS.Host/Jobs/JobQueueWorker.cs index 94f307fcb..ae11e6c2f 100644 --- a/src/FluentCMS.Host/Jobs/JobQueueWorker.cs +++ b/src/FluentCMS.Host/Jobs/JobQueueWorker.cs @@ -8,6 +8,7 @@ namespace FluentCMS.Host.Jobs; public sealed class JobQueueWorker( IServiceScopeFactory scopes, IOptions options, + TimeProvider clock, ILogger logger) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -78,14 +79,16 @@ private async Task 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, 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 +103,22 @@ private async Task 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 +127,99 @@ private async Task 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, TimeSpan duration, CancellationTokenSource running, CancellationToken cancellationToken) + { + var kept = await KeepLeaseAsync( + async renewing => + { + await using var scope = scopes.CreateAsyncScope(); + return await scope.ServiceProvider.GetRequiredService().RenewAsync(job.Id, owner, duration, renewing); + }, + 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 at the next beat while the lease still holds, so one transient error does not give the job away. False once a + // renewal is refused, because another worker holds the lease, or once the lease has run out with no renewal. + public static async Task KeepLeaseAsync( + Func> renew, + TimeSpan duration, + TimeProvider clock, + ILogger logger, + CancellationToken cancellationToken) { + var heldUntil = clock.GetUtcNow() + duration; try { while (true) { await Task.Delay(duration / 3, cancellationToken); - await using var scope = scopes.CreateAsyncScope(); - if (!await scope.ServiceProvider.GetRequiredService().RenewAsync(jobId, owner, duration, cancellationToken)) + var renewedAt = clock.GetUtcNow(); + try { - return; + if (!await renew(cancellationToken)) + { + return false; + } + + heldUntil = renewedAt + duration; + } + catch (Exception exception) when (!cancellationToken.IsCancellationRequested) + { + if (clock.GetUtcNow() >= heldUntil) + { + logger.LogError(exception, "A job's lease ran out at {HeldUntil} while it could not be renewed.", heldUntil); + return false; + } + + logger.LogWarning(exception, "A job's lease could not be renewed; it holds until {HeldUntil} and is tried again.", heldUntil); } } } - catch (OperationCanceledException) + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - // The handler is done. + return true; + } + } + + // The update is guarded by the owner, so a worker that lost the lease meanwhile changes nothing. + private async Task ReleaseAsync(IJobLeases leases, Job job, string owner) + { + try + { + 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; } diff --git a/src/FluentCMS.Infrastructure/Jobs/JobLeases.cs b/src/FluentCMS.Infrastructure/Jobs/JobLeases.cs index 3683daa8c..595eb1ce9 100644 --- a/src/FluentCMS.Infrastructure/Jobs/JobLeases.cs +++ b/src/FluentCMS.Infrastructure/Jobs/JobLeases.cs @@ -135,6 +135,22 @@ public async Task RetryAsync(Guid jobId, string owner, string error, TimeS cancellationToken); } + public Task 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 FinishAsync(Guid jobId, string owner, JobState state, string? error, CancellationToken cancellationToken) { var now = clock.GetUtcNow(); diff --git a/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs b/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs index d8500cf0f..7b3e3e207 100644 --- a/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs +++ b/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using System.Globalization; using System.Text.Json; using FluentCMS.Core.Extensions; using FluentCMS.Core.Identity; @@ -6,7 +7,11 @@ using FluentCMS.Core.Persistence; using FluentCMS.Core.Results; using FluentCMS.Host.Jobs; +using FluentCMS.Infrastructure.Jobs; +using FluentCMS.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; namespace FluentCMS.IntegrationTests.Jobs; @@ -19,6 +24,8 @@ public sealed class JobQueueWorkerTests ["JobQueue:RetryDelay"] = "00:00:00.050", }; + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30); + [Fact] public async Task The_worker_runs_each_job_as_its_initiator_and_records_how_it_ended() { @@ -51,21 +58,179 @@ public async Task The_worker_runs_each_job_as_its_initiator_and_records_how_it_e Assert.Contains("No handler runs jobs of the type 'no-such-type'", orphan.LastError); } - [Fact] - public async Task A_handler_that_works_past_its_lease_without_progress_keeps_the_job_and_runs_once() + // The test moves the clock past the lease twice over, each step only once the heartbeat has renewed the lease at the clock's time, + // so the job stays this worker's whatever the machine's load. + [Theory] + [InlineData(DatabaseProviders.Sqlite)] + [PostgresData] + public async Task A_handler_that_works_past_its_lease_without_progress_keeps_the_job_and_runs_once(string provider) { + // On a whole second, so the lease's end reads back exactly from PostgreSQL, which keeps microseconds. + var clock = new ManualClock(DateTimeOffset.FromUnixTimeSeconds(DateTimeOffset.UtcNow.ToUnixTimeSeconds())); using var install = new TemporaryInstall(); + await using var database = await TemporaryDatabase.CreateAsync(provider); await using var host = await AdminHost.StartAsync( install, - settings: new Dictionary(FastQueue) { ["JobQueue:LeaseDuration"] = "00:00:01.500", ["JobQueue:WorkerCount"] = "3" }, - configure: builder => builder.Services.AddFluentCmsExtension()); + settings: Settings(database, lease: "00:00:03", workers: 3), + configure: builder => builder.Services.AddSingleton(clock).AddFluentCmsExtension()); var administrator = await host.CreateFirstAdministratorAsync(); + var job = await EnqueueAsync(host, administrator, new NewJob(WaitingHandler.Name, default)); + var run = WaitingHandler.For(job.Id); + await run.Started.Task.WaitAsync(Timeout); + + foreach (var _ in Enumerable.Range(0, 4)) + { + clock.Now += TimeSpan.FromSeconds(2); + await UntilAsync(async () => (await JobRowAsync(host, job.Id)).LeaseUntil >= clock.Now + TimeSpan.FromSeconds(3)); + } - var job = await EnqueueAsync(host, administrator, new NewJob(SlowHandler.Name, default)); + run.Finish.SetResult(); var finished = await FinishedAsync(host, administrator, job.Id); Assert.Equal((JobState.Succeeded, 1), (finished.State, finished.Attempts)); - Assert.Equal(1, SlowHandler.Runs[job.Id]); + Assert.Equal(1, run.Runs); + } + + // The clock stands still, so only the stolen lease can stop the handler; the next job shows the worker has moved on. + [Theory] + [InlineData(DatabaseProviders.Sqlite)] + [PostgresData] + public async Task A_handler_whose_lease_another_worker_took_is_stopped_and_its_worker_records_nothing(string provider) + { + using var install = new TemporaryInstall(); + await using var database = await TemporaryDatabase.CreateAsync(provider); + await using var host = await AdminHost.StartAsync( + install, + settings: Settings(database), + configure: builder => builder.Services + .AddSingleton(new ManualClock(DateTimeOffset.UtcNow)) + .AddFluentCmsExtension() + .AddFluentCmsExtension()); + var administrator = await host.CreateFirstAdministratorAsync(); + var job = await EnqueueAsync(host, administrator, new NewJob(WaitingHandler.Name, default)); + var run = WaitingHandler.For(job.Id); + await run.Started.Task.WaitAsync(Timeout); + + await QueryAsync(host, context => context.Jobs + .Where(row => row.Id == job.Id) + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.LeaseOwner, "another-worker"))); + await run.Cancelled.Task.WaitAsync(Timeout); + var next = await EnqueueAsync(host, administrator, new NewJob(CountingHandler.Name, JsonDocument.Parse("""{"items": 1}""").RootElement.Clone())); + Assert.Equal(JobState.Succeeded, (await FinishedAsync(host, administrator, next.Id)).State); + + var stolen = await JobRowAsync(host, job.Id); + Assert.Equal((JobState.Running, "another-worker", 1), (stolen.State, stolen.LeaseOwner, stolen.Attempts)); + Assert.Null(stolen.LastError); + Assert.Null(stolen.FinishedAt); + Assert.Equal(1, run.Runs); + } + + // A job allowed one attempt shows the attempt was given back: without that, the restart would find it dead. + [Theory] + [InlineData(DatabaseProviders.Sqlite)] + [PostgresData] + public async Task A_job_running_when_the_host_stops_is_queued_again_with_its_attempt_and_runs_after_a_restart(string provider) + { + using var install = new TemporaryInstall(); + await using var database = await TemporaryDatabase.CreateAsync(provider); + Guid administrator; + Job job; + WaitingHandler.Run run; + await using (var host = await AdminHost.StartAsync(install, settings: Settings(database), configure: AddWaitingHandler)) + { + administrator = await host.CreateFirstAdministratorAsync(); + job = await EnqueueAsync(host, administrator, new NewJob(WaitingHandler.Name, default) { MaxAttempts = 1 }); + run = WaitingHandler.For(job.Id); + await run.Started.Task.WaitAsync(Timeout); + + await host.App.StopAsync(); + + await run.Cancelled.Task.WaitAsync(Timeout); + var released = await JobRowAsync(host, job.Id); + Assert.Equal((JobState.Queued, 0, (string?)null, (DateTimeOffset?)null), (released.State, released.Attempts, released.LeaseOwner, released.LeaseUntil)); + } + + run.Finish.SetResult(); + await using var restarted = await AdminHost.StartAsync(install, settings: Settings(database), configure: AddWaitingHandler); + var finished = await FinishedAsync(restarted, administrator, job.Id); + + Assert.Equal((JobState.Succeeded, 1), (finished.State, finished.Attempts)); + Assert.Equal(2, run.Runs); + } + + [Fact] + public async Task A_renewal_that_throws_is_tried_again_at_the_next_beat() + { + var calls = 0; + var third = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var done = new CancellationTokenSource(); + + var keeping = JobQueueWorker.KeepLeaseAsync( + _ => + { + var call = Interlocked.Increment(ref calls); + if (call == 3) + { + third.TrySetResult(); + } + + return call == 1 ? Task.FromException(new TimeoutException("The database did not answer.")) : Task.FromResult(true); + }, + TimeSpan.FromMilliseconds(300), + new ManualClock(DateTimeOffset.UtcNow), + NullLogger.Instance, + done.Token); + await third.Task.WaitAsync(Timeout); + await done.CancelAsync(); + + Assert.True(await keeping.WaitAsync(Timeout)); + } + + // The clock stands still while renewals fail, so the lease holds however many beats pass; moving it past the lease ends it. + [Fact] + public async Task Renewals_that_keep_failing_give_the_lease_up_once_it_has_run_out_and_not_before() + { + var clock = new ManualClock(DateTimeOffset.UtcNow); + var lease = TimeSpan.FromMilliseconds(300); + var failures = 0; + var third = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var keeping = JobQueueWorker.KeepLeaseAsync( + _ => + { + if (Interlocked.Increment(ref failures) == 3) + { + third.TrySetResult(); + } + + throw new TimeoutException("The database did not answer."); + }, + lease, + clock, + NullLogger.Instance, + CancellationToken.None); + await third.Task.WaitAsync(Timeout); + + Assert.False(keeping.IsCompleted); + + clock.Now += lease; + + Assert.False(await keeping.WaitAsync(Timeout)); + } + + [Fact] + public async Task A_refused_renewal_gives_the_lease_up_at_once() + { + var calls = 0; + + var kept = await JobQueueWorker.KeepLeaseAsync( + _ => Task.FromResult(Interlocked.Increment(ref calls) > 1), + TimeSpan.FromMilliseconds(300), + new ManualClock(DateTimeOffset.UtcNow), + NullLogger.Instance, + CancellationToken.None).WaitAsync(Timeout); + + Assert.Equal((false, 1), (kept, calls)); } [Fact] @@ -78,6 +243,50 @@ public void The_retry_delay_doubles_with_each_attempt_up_to_its_ceiling() Enumerable.Range(1, 5).Select(attempt => JobQueueWorker.RetryDelay(settings, attempt))); } + [Fact] + public void The_retry_delay_stops_at_its_ceiling_however_many_attempts_there_were() + { + var settings = new JobQueueOptions { RetryDelay = TimeSpan.FromSeconds(30), MaxRetryDelay = TimeSpan.FromHours(1) }; + + Assert.All(new[] { 36, 1_000, int.MaxValue }, attempt => Assert.Equal(TimeSpan.FromHours(1), JobQueueWorker.RetryDelay(settings, attempt))); + Assert.Equal(TimeSpan.Zero, JobQueueWorker.RetryDelay(new JobQueueOptions { RetryDelay = TimeSpan.Zero }, int.MaxValue)); + } + + private static Dictionary Settings(TemporaryDatabase database, string lease = "00:00:01.500", int workers = 1) + { + return new Dictionary(database.Configuration.Concat(FastQueue)) + { + ["JobQueue:LeaseDuration"] = lease, + ["JobQueue:WorkerCount"] = workers.ToString(CultureInfo.InvariantCulture), + }; + } + + private static void AddWaitingHandler(Microsoft.AspNetCore.Builder.WebApplicationBuilder builder) + { + builder.Services.AddFluentCmsExtension(); + } + + private static async Task QueryAsync(AdminHost host, Func> query) + { + await using var scope = host.App.Services.CreateAsyncScope(); + return await query(scope.ServiceProvider.GetRequiredService()); + } + + private static Task JobRowAsync(AdminHost host, Guid id) + { + return QueryAsync(host, context => context.Jobs.AsNoTracking().SingleAsync(row => row.Id == id)); + } + + private static async Task UntilAsync(Func> condition) + { + var deadline = DateTime.UtcNow + Timeout; + while (!await condition()) + { + Assert.True(DateTime.UtcNow < deadline, "The condition did not hold in time."); + await Task.Delay(50); + } + } + private static Task EnqueueAsync(AdminHost host, Guid administrator, NewJob job) { return host.AsUserAsync(administrator, async services => @@ -138,20 +347,55 @@ public Task RunAsync(JobContext context, CancellationToken cancellationT } } - private sealed class SlowHandler : IJobHandler + // Waits until the test lets it finish or its token is cancelled, and records which it was. + private sealed class WaitingHandler : IJobHandler { - public const string Name = "test.slow"; + public const string Name = "test.waiting"; - public static ConcurrentDictionary Runs { get; } = new(); + private static readonly ConcurrentDictionary Runs = new(); public string Type => Name; + public static Run For(Guid jobId) + { + return Runs.GetOrAdd(jobId, _ => new Run()); + } + public async Task RunAsync(JobContext context, CancellationToken cancellationToken) { - Runs.AddOrUpdate(context.Job.Id, 1, (_, runs) => runs + 1); - await Task.Delay(TimeSpan.FromSeconds(3.5), cancellationToken); + var run = For(context.Job.Id); + run.Count(); + run.Started.TrySetResult(); + try + { + await run.Finish.Task.WaitAsync(cancellationToken); + } + catch (OperationCanceledException) + { + run.Cancelled.TrySetResult(); + throw; + } + return Result.Success(); } + + public sealed class Run + { + private int _runs; + + public int Runs => _runs; + + public TaskCompletionSource Started { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public TaskCompletionSource Finish { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public TaskCompletionSource Cancelled { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public void Count() + { + Interlocked.Increment(ref _runs); + } + } } private sealed class ThrowingHandler : IJobHandler From c1dbcc007597f92df48b7dbbacf7879c048a89ed Mon Sep 17 00:00:00 2001 From: Amir Pournasserian Date: Mon, 19 Oct 2026 03:15:15 -0400 Subject: [PATCH 2/3] End the lease heartbeat at the lease's end, and count a renewal after the handler as done --- CONTRIBUTING.md | 2 +- src/FluentCMS.Host/Jobs/JobQueueWorker.cs | 72 ++++++----- .../Jobs/JobQueueWorkerTests.cs | 114 ++++++++++++++++-- 3 files changed, 150 insertions(+), 38 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index cb202bf24..dd8884808 100644 --- a/CONTRIBUTING.md +++ b/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 at a third of it while the handler runs. A renewal that fails is tried again while the lease holds. Once the lease is lost, or runs out with no renewal, the handler's cancellation token fires, because another worker may 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. +`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 diff --git a/src/FluentCMS.Host/Jobs/JobQueueWorker.cs b/src/FluentCMS.Host/Jobs/JobQueueWorker.cs index ae11e6c2f..56b641ecb 100644 --- a/src/FluentCMS.Host/Jobs/JobQueueWorker.cs +++ b/src/FluentCMS.Host/Jobs/JobQueueWorker.cs @@ -56,6 +56,8 @@ private async Task RunNextAsync(string owner, JobQueueOptions settings, Ca await using var leaseScope = scopes.CreateAsyncScope(); var leases = leaseScope.ServiceProvider.GetRequiredService(); + // 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) { @@ -83,7 +85,7 @@ private async Task RunNextAsync(string owner, JobQueueOptions settings, Ca // 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 = KeepAsync(job, owner, settings.LeaseDuration, running, 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 @@ -132,7 +134,7 @@ private async Task RunNextAsync(string owner, JobQueueOptions settings, Ca } // Each renewal has a scope of its own; a lost lease stops the handler. - private async Task KeepAsync(Job job, string owner, TimeSpan duration, CancellationTokenSource running, CancellationToken cancellationToken) + private async Task KeepAsync(Job job, string owner, DateTimeOffset heldUntil, TimeSpan duration, CancellationTokenSource running, CancellationToken cancellationToken) { var kept = await KeepLeaseAsync( async renewing => @@ -140,6 +142,7 @@ private async Task KeepAsync(Job job, string owner, TimeSpan duration, Cancellat await using var scope = scopes.CreateAsyncScope(); return await scope.ServiceProvider.GetRequiredService().RenewAsync(job.Id, owner, duration, renewing); }, + heldUntil, duration, clock, logger, @@ -153,46 +156,57 @@ private async Task KeepAsync(Job job, string owner, TimeSpan duration, Cancellat } // 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 at the next beat while the lease still holds, so one transient error does not give the job away. False once a - // renewal is refused, because another worker holds the lease, or once the lease has run out with no renewal. + // 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 KeepLeaseAsync( Func> renew, + DateTimeOffset heldUntil, TimeSpan duration, TimeProvider clock, ILogger logger, CancellationToken cancellationToken) { - var heldUntil = clock.GetUtcNow() + duration; - try + while (true) { - 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(duration / 3, cancellationToken); - var renewedAt = clock.GetUtcNow(); - try - { - if (!await renew(cancellationToken)) - { - return false; - } + await Task.Delay(wait > TimeSpan.Zero ? wait : TimeSpan.Zero, clock, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + return true; + } - heldUntil = renewedAt + duration; - } - catch (Exception exception) when (!cancellationToken.IsCancellationRequested) - { - if (clock.GetUtcNow() >= heldUntil) - { - logger.LogError(exception, "A job's lease ran out at {HeldUntil} while it could not be renewed.", heldUntil); - return false; - } + var renewingAt = clock.GetUtcNow(); + if (renewingAt >= heldUntil) + { + logger.LogError("A job's lease ran out at {HeldUntil} with no renewal.", heldUntil); + return false; + } - logger.LogWarning(exception, "A job's lease could not be renewed; it holds until {HeldUntil} and is tried again.", heldUntil); + using var attempt = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + attempt.CancelAfter(heldUntil - renewingAt); + try + { + if (!await renew(attempt.Token)) + { + 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) when (cancellationToken.IsCancellationRequested) - { - return true; } } diff --git a/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs b/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs index 7b3e3e207..38fd42a93 100644 --- a/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs +++ b/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using System.Diagnostics; using System.Globalization; using System.Text.Json; using FluentCMS.Core.Extensions; @@ -91,6 +92,46 @@ public async Task A_handler_that_works_past_its_lease_without_progress_keeps_the Assert.Equal(1, run.Runs); } + // A trigger makes the renewals fail while the clock stands still, so the lease holds; once they work again, the clock moves past + // the lease twice over and the job is still this worker's. + [Fact] + public async Task A_running_worker_keeps_the_job_through_renewals_that_fail_and_runs_it_once() + { + var clock = new ManualClock(DateTimeOffset.FromUnixTimeSeconds(DateTimeOffset.UtcNow.ToUnixTimeSeconds())); + using var install = new TemporaryInstall(); + await using var host = await AdminHost.StartAsync( + install, + settings: new Dictionary(FastQueue) { ["JobQueue:LeaseDuration"] = "00:00:00.600", ["JobQueue:WorkerCount"] = "2" }, + configure: builder => builder.Services.AddSingleton(clock).AddFluentCmsExtension()); + var administrator = await host.CreateFirstAdministratorAsync(); + await QueryAsync(host, context => context.Database.ExecuteSqlRawAsync(""" + CREATE TABLE "FailRenewals" ("Id" INTEGER PRIMARY KEY); + INSERT INTO "FailRenewals" ("Id") VALUES (1); + CREATE TRIGGER "FailRenewal" BEFORE UPDATE OF "LeaseUntil" ON "Jobs" + WHEN OLD."State" = 'Running' AND NEW."State" = 'Running' AND EXISTS (SELECT 1 FROM "FailRenewals") + BEGIN SELECT RAISE(ABORT, 'The database is unreachable.'); END; + """)); + + var job = await EnqueueAsync(host, administrator, new NewJob(WaitingHandler.Name, default)); + var run = WaitingHandler.For(job.Id); + await run.Started.Task.WaitAsync(Timeout); + await Task.Delay(TimeSpan.FromSeconds(1)); + await QueryAsync(host, context => context.Database.ExecuteSqlRawAsync("""DELETE FROM "FailRenewals";""")); + + foreach (var _ in Enumerable.Range(0, 4)) + { + clock.Now += TimeSpan.FromMilliseconds(400); + await UntilAsync(async () => (await JobRowAsync(host, job.Id)).LeaseUntil >= clock.Now + TimeSpan.FromMilliseconds(600)); + } + + run.Finish.SetResult(); + var finished = await FinishedAsync(host, administrator, job.Id); + + Assert.Equal((JobState.Succeeded, 1), (finished.State, finished.Attempts)); + Assert.Equal(1, run.Runs); + Assert.False(run.Cancelled.Task.IsCompleted); + } + // The clock stands still, so only the stolen lease can stop the handler; the next job shows the worker has moved on. [Theory] [InlineData(DatabaseProviders.Sqlite)] @@ -148,6 +189,7 @@ public async Task A_job_running_when_the_host_stops_is_queued_again_with_its_att await run.Cancelled.Task.WaitAsync(Timeout); var released = await JobRowAsync(host, job.Id); Assert.Equal((JobState.Queued, 0, (string?)null, (DateTimeOffset?)null), (released.State, released.Attempts, released.LeaseOwner, released.LeaseUntil)); + Assert.True(released.ScheduledAt <= DateTimeOffset.UtcNow, "The released job is not due at once."); } run.Finish.SetResult(); @@ -161,6 +203,8 @@ public async Task A_job_running_when_the_host_stops_is_queued_again_with_its_att [Fact] public async Task A_renewal_that_throws_is_tried_again_at_the_next_beat() { + var clock = new ManualClock(DateTimeOffset.UtcNow); + var lease = TimeSpan.FromMilliseconds(300); var calls = 0; var third = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using var done = new CancellationTokenSource(); @@ -176,8 +220,9 @@ public async Task A_renewal_that_throws_is_tried_again_at_the_next_beat() return call == 1 ? Task.FromException(new TimeoutException("The database did not answer.")) : Task.FromResult(true); }, - TimeSpan.FromMilliseconds(300), - new ManualClock(DateTimeOffset.UtcNow), + clock.Now + lease, + lease, + clock, NullLogger.Instance, done.Token); await third.Task.WaitAsync(Timeout); @@ -186,9 +231,10 @@ public async Task A_renewal_that_throws_is_tried_again_at_the_next_beat() Assert.True(await keeping.WaitAsync(Timeout)); } - // The clock stands still while renewals fail, so the lease holds however many beats pass; moving it past the lease ends it. + // The clock stands still while renewals fail, so the lease holds however many beats pass. Once it is past the deadline, the + // lease is given up without trying another renewal. [Fact] - public async Task Renewals_that_keep_failing_give_the_lease_up_once_it_has_run_out_and_not_before() + public async Task Renewals_that_keep_failing_give_the_lease_up_at_its_deadline_and_not_before() { var clock = new ManualClock(DateTimeOffset.UtcNow); var lease = TimeSpan.FromMilliseconds(300); @@ -205,34 +251,86 @@ public async Task Renewals_that_keep_failing_give_the_lease_up_once_it_has_run_o throw new TimeoutException("The database did not answer."); }, + clock.Now + lease, lease, clock, NullLogger.Instance, CancellationToken.None); await third.Task.WaitAsync(Timeout); - Assert.False(keeping.IsCompleted); - clock.Now += lease; + var before = clock.Now; + clock.Now = before + lease; + var tried = Volatile.Read(ref failures); Assert.False(await keeping.WaitAsync(Timeout)); + Assert.InRange(failures, tried, tried + 1); + } + + // On the real clock: a renewal that hangs is cut off at the deadline, so the lease is given up then and not a beat or a timeout later. + [Fact] + public async Task A_renewal_that_hangs_gives_the_lease_up_at_its_deadline() + { + var lease = TimeSpan.FromMilliseconds(600); + var started = Stopwatch.StartNew(); + + var kept = await JobQueueWorker.KeepLeaseAsync( + async renewing => + { + await Task.Delay(System.Threading.Timeout.InfiniteTimeSpan, renewing); + return true; + }, + DateTimeOffset.UtcNow + lease, + lease, + TimeProvider.System, + NullLogger.Instance, + CancellationToken.None).WaitAsync(Timeout); + + Assert.False(kept); + Assert.True(started.Elapsed >= lease - TimeSpan.FromMilliseconds(50), $"Gave up after {started.Elapsed}, before the lease ran out."); } [Fact] public async Task A_refused_renewal_gives_the_lease_up_at_once() { + var clock = new ManualClock(DateTimeOffset.UtcNow); + var lease = TimeSpan.FromMilliseconds(300); var calls = 0; var kept = await JobQueueWorker.KeepLeaseAsync( _ => Task.FromResult(Interlocked.Increment(ref calls) > 1), - TimeSpan.FromMilliseconds(300), - new ManualClock(DateTimeOffset.UtcNow), + clock.Now + lease, + lease, + clock, NullLogger.Instance, CancellationToken.None).WaitAsync(Timeout); Assert.Equal((false, 1), (kept, calls)); } + // Once the handler is done, how the renewal in flight ended no longer matters, so it never stops the outcome being recorded. + [Fact] + public async Task A_renewal_that_fails_as_the_handler_finishes_counts_as_done() + { + var clock = new ManualClock(DateTimeOffset.UtcNow); + var lease = TimeSpan.FromMilliseconds(300); + using var done = new CancellationTokenSource(); + + var kept = await JobQueueWorker.KeepLeaseAsync( + _ => + { + done.Cancel(); + throw new TimeoutException("The database did not answer."); + }, + clock.Now + lease, + lease, + clock, + NullLogger.Instance, + done.Token).WaitAsync(Timeout); + + Assert.True(kept); + } + [Fact] public void The_retry_delay_doubles_with_each_attempt_up_to_its_ceiling() { From 38b497d43477aca0c85e2ad9b616dd50dfb6af8f Mon Sep 17 00:00:00 2001 From: Amir Pournasserian Date: Mon, 19 Oct 2026 04:11:37 -0400 Subject: [PATCH 3/3] Step the test clock by less than half the lease --- .../Jobs/JobQueueWorkerTests.cs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs b/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs index 38fd42a93..0bdc8e49f 100644 --- a/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs +++ b/tests/FluentCMS.IntegrationTests/Jobs/JobQueueWorkerTests.cs @@ -79,9 +79,10 @@ public async Task A_handler_that_works_past_its_lease_without_progress_keeps_the var run = WaitingHandler.For(job.Id); await run.Started.Task.WaitAsync(Timeout); - foreach (var _ in Enumerable.Range(0, 4)) + // Steps under half the lease, so a renewal that read the clock just before a step still holds past the next one. + foreach (var _ in Enumerable.Range(0, 8)) { - clock.Now += TimeSpan.FromSeconds(2); + clock.Now += TimeSpan.FromSeconds(1); await UntilAsync(async () => (await JobRowAsync(host, job.Id)).LeaseUntil >= clock.Now + TimeSpan.FromSeconds(3)); } @@ -118,9 +119,9 @@ CREATE TRIGGER "FailRenewal" BEFORE UPDATE OF "LeaseUntil" ON "Jobs" await Task.Delay(TimeSpan.FromSeconds(1)); await QueryAsync(host, context => context.Database.ExecuteSqlRawAsync("""DELETE FROM "FailRenewals";""")); - foreach (var _ in Enumerable.Range(0, 4)) + foreach (var _ in Enumerable.Range(0, 6)) { - clock.Now += TimeSpan.FromMilliseconds(400); + clock.Now += TimeSpan.FromMilliseconds(250); await UntilAsync(async () => (await JobRowAsync(host, job.Id)).LeaseUntil >= clock.Now + TimeSpan.FromMilliseconds(600)); }