From 4793ad183287168dda5596f43dd8a145afaa1d6e Mon Sep 17 00:00:00 2001 From: xiong Date: Tue, 4 Aug 2026 14:29:11 +0800 Subject: [PATCH] fix(worker): enable safe multi-instance job processing --- .../Jobs/BackgroundJobService.cs | 6 +- .../Processor/BackgroundJobLeaseManager.cs | 165 ++++++++++++++++++ .../BackgroundJobService.Processor.cs | 117 +++++++++---- Tiku.Infrastructure/Modules/JobsModule.cs | 4 + .../Observability/WorkerTelemetry.cs | 10 +- .../Properties/AssemblyInfo.cs | 1 + Tiku.IntegrationTests/Api/ApiTestFactory.cs | 7 +- .../Api/MonolithBackgroundProcessingTests.cs | 136 ++++++++++++++- .../DockerPostgresTestDefaults.cs | 28 +++ .../Infrastructure/PostgresTestDatabase.cs | 4 +- .../ModulePersistenceBoundaryTests.cs | 4 +- .../PersistenceModelTests.cs | 5 +- Tiku.UnitTests/Tiku.UnitTests.csproj | 1 + .../PeriodicWorkerExecutionModeTests.cs | 76 ++++++++ Tiku.Worker/Tiku.Worker.csproj | 4 + Tiku.Worker/WorkerServices.cs | 55 +++++- 16 files changed, 570 insertions(+), 53 deletions(-) create mode 100644 Tiku.Infrastructure/Jobs/Processor/BackgroundJobLeaseManager.cs create mode 100644 Tiku.IntegrationTests/Infrastructure/DockerPostgresTestDefaults.cs create mode 100644 Tiku.UnitTests/Worker/PeriodicWorkerExecutionModeTests.cs diff --git a/Tiku.Infrastructure/Jobs/BackgroundJobService.cs b/Tiku.Infrastructure/Jobs/BackgroundJobService.cs index 457da05..79be704 100644 --- a/Tiku.Infrastructure/Jobs/BackgroundJobService.cs +++ b/Tiku.Infrastructure/Jobs/BackgroundJobService.cs @@ -7,9 +7,11 @@ namespace Tiku.Infrastructure.Jobs; internal sealed partial class BackgroundJobService( IJobsOperationsPersistence dbContext, ITenantExecutionScope tenantExecutionScope, - IFeatureAccessService featureAccessService) : IBackgroundJobService + IFeatureAccessService featureAccessService, + BackgroundJobLeaseManager leaseManager, + BackgroundJobLeaseTiming leaseTiming) : IBackgroundJobService { private IJobsOperationsPersistence jobsOperationsPersistence { get; } = dbContext; private IModulePersistence unitOfWork { get; } = dbContext; - private static readonly TimeSpan LeaseDuration = TimeSpan.FromMinutes(5); + private TimeSpan LeaseDuration => leaseTiming.LeaseDuration; } diff --git a/Tiku.Infrastructure/Jobs/Processor/BackgroundJobLeaseManager.cs b/Tiku.Infrastructure/Jobs/Processor/BackgroundJobLeaseManager.cs new file mode 100644 index 0000000..46f32a7 --- /dev/null +++ b/Tiku.Infrastructure/Jobs/Processor/BackgroundJobLeaseManager.cs @@ -0,0 +1,165 @@ +using Microsoft.Extensions.Logging; +using Npgsql; +using Tiku.Infrastructure.Observability; + +namespace Tiku.Infrastructure.Jobs; + +internal sealed record BackgroundJobLeaseTiming( + TimeSpan LeaseDuration, + TimeSpan RenewalInterval, + TimeSpan RetryInterval) +{ + internal static readonly BackgroundJobLeaseTiming Default = new( + TimeSpan.FromMinutes(5), + TimeSpan.FromMinutes(1), + TimeSpan.FromSeconds(10)); +} + +internal sealed class BackgroundJobLeaseManager( + NpgsqlDataSource dataSource, + BackgroundJobLeaseTiming timing, + TimeProvider timeProvider, + ILogger logger) +{ + internal BackgroundJobLease Start( + Guid jobId, + string workerId, + DateTimeOffset leaseExpiresAt, + CancellationToken executionCancellationToken) + { + return new BackgroundJobLease( + jobId, + workerId, + leaseExpiresAt, + dataSource, + timing, + timeProvider, + logger, + executionCancellationToken); + } +} + +internal sealed class BackgroundJobLease : IAsyncDisposable +{ + private readonly Guid jobId; + private readonly string workerId; + private readonly NpgsqlDataSource dataSource; + private readonly BackgroundJobLeaseTiming timing; + private readonly TimeProvider timeProvider; + private readonly ILogger logger; + private readonly CancellationTokenSource executionCancellation; + private readonly CancellationTokenSource renewalCancellation = new(); + private readonly Task renewalTask; + private DateTimeOffset confirmedExpiresAt; + private int ownershipLost; + private int stopped; + + internal BackgroundJobLease( + Guid jobId, + string workerId, + DateTimeOffset leaseExpiresAt, + NpgsqlDataSource dataSource, + BackgroundJobLeaseTiming timing, + TimeProvider timeProvider, + ILogger logger, + CancellationToken executionCancellationToken) + { + this.jobId = jobId; + this.workerId = workerId; + this.dataSource = dataSource; + this.timing = timing; + this.timeProvider = timeProvider; + this.logger = logger; + confirmedExpiresAt = leaseExpiresAt; + executionCancellation = CancellationTokenSource.CreateLinkedTokenSource(executionCancellationToken); + renewalTask = RenewAsync(); + } + + internal CancellationToken ExecutionToken => executionCancellation.Token; + internal bool OwnershipLost => Volatile.Read(ref ownershipLost) != 0; + + internal async Task StopAsync() + { + if (Interlocked.Exchange(ref stopped, 1) != 0) return; + + await renewalCancellation.CancelAsync(); + try + { + await renewalTask; + } + catch (OperationCanceledException) when (renewalCancellation.IsCancellationRequested) + { + } + } + + public async ValueTask DisposeAsync() + { + await StopAsync(); + renewalCancellation.Dispose(); + executionCancellation.Dispose(); + } + + private async Task RenewAsync() + { + var delay = timing.RenewalInterval; + while (!renewalCancellation.IsCancellationRequested) + { + await Task.Delay(delay, timeProvider, renewalCancellation.Token); + try + { + var now = timeProvider.GetUtcNow(); + var nextExpiresAt = now.Add(timing.LeaseDuration); + await using var connection = await dataSource.OpenConnectionAsync(renewalCancellation.Token); + await using var command = connection.CreateCommand(); + command.CommandText = """ + UPDATE background_jobs + SET lock_expires_at = @lock_expires_at, + updated_at = @updated_at + WHERE id = @id + AND locked_by = @worker_id + AND status = 'processing' + """; + command.Parameters.AddWithValue("lock_expires_at", nextExpiresAt); + command.Parameters.AddWithValue("updated_at", now); + command.Parameters.AddWithValue("id", jobId); + command.Parameters.AddWithValue("worker_id", workerId); + var renewed = await command.ExecuteNonQueryAsync(renewalCancellation.Token); + if (renewed == 0) + { + LoseOwnership("The job is no longer owned by this worker."); + return; + } + + confirmedExpiresAt = nextExpiresAt; + delay = timing.RenewalInterval; + WorkerTelemetry.RecordJobLeaseRenewal("succeeded"); + } + catch (OperationCanceledException) when (renewalCancellation.IsCancellationRequested) + { + return; + } + catch (Exception exception) + { + WorkerTelemetry.RecordJobLeaseRenewal("failed"); + logger.LogWarning(exception, "Failed to renew background job {JobId} lease for {WorkerId}.", jobId, + workerId); + if (timeProvider.GetUtcNow() >= confirmedExpiresAt) + { + LoseOwnership("The last confirmed lease expired before it could be renewed."); + return; + } + + delay = timing.RetryInterval; + } + } + } + + private void LoseOwnership(string reason) + { + if (Interlocked.Exchange(ref ownershipLost, 1) != 0) return; + + WorkerTelemetry.RecordJobLeaseRenewal("lost"); + logger.LogError("Background job {JobId} lease was lost by {WorkerId}. {Reason}", jobId, workerId, reason); + executionCancellation.Cancel(); + } +} diff --git a/Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs b/Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs index 8a6df40..047aed1 100644 --- a/Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs +++ b/Tiku.Infrastructure/Jobs/Processor/BackgroundJobService.Processor.cs @@ -17,6 +17,30 @@ internal sealed partial class BackgroundJobService int batchSize, bool includeImmediateJobs = true, CancellationToken cancellationToken = default) + { + var processed = 0; + var limit = Math.Clamp(batchSize, 1, 100); + for (var index = 0; index < limit; index++) + { + cancellationToken.ThrowIfCancellationRequested(); + var jobId = await TryClaimNextAsync(workerId, includeImmediateJobs, cancellationToken); + if (!jobId.HasValue) break; + + unitOfWork.ChangeTracker.Clear(); + var job = await jobsOperationsPersistence.BackgroundJobs.SingleAsync( + value => value.Id == jobId.Value, + cancellationToken); + if (await ProcessJobAsync(job, workerId, true, cancellationToken)) processed++; + unitOfWork.ChangeTracker.Clear(); + } + + return processed; + } + + private async Task TryClaimNextAsync( + string workerId, + bool includeImmediateJobs, + CancellationToken cancellationToken) { var now = DateTimeOffset.UtcNow; var leaseExpiresAt = now.Add(LeaseDuration); @@ -27,7 +51,7 @@ internal sealed partial class BackgroundJobService lock_expires_at = {leaseExpiresAt}, started_at = COALESCE(started_at, {now}), updated_at = {now} - WHERE job.id IN ( + WHERE job.id = ( SELECT candidate.id FROM background_jobs AS candidate WHERE ( @@ -37,23 +61,12 @@ internal sealed partial class BackgroundJobService ) ORDER BY candidate.created_at, candidate.id FOR UPDATE SKIP LOCKED - LIMIT {Math.Clamp(batchSize, 1, 100)} + LIMIT 1 ) RETURNING job.id AS "Value" """) .ToArrayAsync(cancellationToken); - - var processed = 0; - unitOfWork.ChangeTracker.Clear(); - foreach (var jobId in claimedIds) - { - cancellationToken.ThrowIfCancellationRequested(); - var job = await jobsOperationsPersistence.BackgroundJobs.SingleAsync(value => value.Id == jobId, cancellationToken); - if (await ProcessJobAsync(job, workerId, true, cancellationToken)) processed++; - unitOfWork.ChangeTracker.Clear(); - } - - return processed; + return claimedIds.Length == 0 ? null : claimedIds[0]; } public async Task ProcessRequestedAsync( @@ -97,17 +110,18 @@ internal sealed partial class BackgroundJobService (alreadyClaimed && (job.Status != BackgroundJobStatus.Processing || job.LockedBy != workerId))) return false; await unitOfWork.Entry(job).ReloadAsync(cancellationToken); + if (alreadyClaimed && (job.Status != BackgroundJobStatus.Processing || job.LockedBy != workerId)) + return false; if (job.CancellationRequestedAt.HasValue) { job.CompletedAt = DateTimeOffset.UtcNow; - await CompleteAsync( + return await CompleteAsync( job, workerId, BackgroundJobStatus.Cancelled, job.Result, null, - cancellationToken); - return true; + cancellationToken) > 0; } if (job.JobType is not ("asset_security_scan" or "tenant_export") && !(await featureAccessService.EvaluateAsync( @@ -116,14 +130,13 @@ internal sealed partial class BackgroundJobService FeatureAccessOperation.Write, cancellationToken)).Allowed) { - await CompleteAsync( + return await CompleteAsync( job, workerId, BackgroundJobStatus.Failed, JsonDefaults.Object(), "Tenant feature entitlement was revoked before job execution.", - cancellationToken); - return true; + cancellationToken) > 0; } if (!alreadyClaimed) @@ -136,6 +149,13 @@ internal sealed partial class BackgroundJobService await unitOfWork.SaveChangesAsync(cancellationToken); } + if (!job.LockExpiresAt.HasValue) return false; + + await using var lease = leaseManager.Start( + job.Id, + workerId, + job.LockExpiresAt.Value, + cancellationToken); try { var handlerResult = await tenantExecutionScope.ExecuteAsync( @@ -150,7 +170,13 @@ internal sealed partial class BackgroundJobService new BackgroundJobExecutionContext(job.Id, job.TenantId, job.JobType, job.Payload), token); }, - cancellationToken); + lease.ExecutionToken); + if (lease.OwnershipLost) + { + RecordLeaseLost(job, startedTimestamp); + return false; + } + var cancellationRequested = await jobsOperationsPersistence.BackgroundJobs.AsNoTracking() .Where(item => item.Id == job.Id) .Select(item => item.CancellationRequestedAt != null) @@ -161,6 +187,16 @@ internal sealed partial class BackgroundJobService job.Result = handlerResult.Result; job.OutputAssetId = handlerResult.OutputAssetId; } + catch (OperationCanceledException) when (lease.OwnershipLost) + { + RecordLeaseLost(job, startedTimestamp); + return false; + } + catch (Exception) when (lease.OwnershipLost) + { + RecordLeaseLost(job, startedTimestamp); + return false; + } catch (Exception exception) when (exception is not OperationCanceledException) { job.RetryCount++; @@ -188,24 +224,39 @@ internal sealed partial class BackgroundJobService exception, token); }, - cancellationToken); + lease.ExecutionToken); } catch (Exception compensationException) when (compensationException is not OperationCanceledException) { job.LastError = $"{job.LastError} Compensation failed: {compensationException.Message}"; } - } - finally - { - await CompleteAsync(job, workerId, job.Status, job.Result, job.LastError, cancellationToken); - WorkerTelemetry.RecordJob(job.JobType, job.Status.ToString(), - Stopwatch.GetElapsedTime(startedTimestamp).TotalMilliseconds); + catch (OperationCanceledException) when (lease.OwnershipLost) + { + RecordLeaseLost(job, startedTimestamp); + return false; + } } + await lease.StopAsync(); + if (lease.OwnershipLost) + { + RecordLeaseLost(job, startedTimestamp); + return false; + } + + var completed = await CompleteAsync(job, workerId, job.Status, job.Result, job.LastError, cancellationToken); + if (completed == 0) + { + RecordLeaseLost(job, startedTimestamp); + return false; + } + + WorkerTelemetry.RecordJob(job.JobType, job.Status.ToString(), + Stopwatch.GetElapsedTime(startedTimestamp).TotalMilliseconds); return true; } - private async Task CompleteAsync( + private async Task CompleteAsync( BackgroundJob job, string workerId, BackgroundJobStatus status, @@ -213,7 +264,7 @@ internal sealed partial class BackgroundJobService string? lastError, CancellationToken cancellationToken) { - await jobsOperationsPersistence.BackgroundJobs + return await jobsOperationsPersistence.BackgroundJobs .Where(value => value.Id == job.Id && value.LockedBy == workerId) .ExecuteUpdateAsync(setters => setters .SetProperty(value => value.Status, status) @@ -227,4 +278,10 @@ internal sealed partial class BackgroundJobService .SetProperty(value => value.LockExpiresAt, (DateTimeOffset?)null), cancellationToken); } + + private static void RecordLeaseLost(BackgroundJob job, long startedTimestamp) + { + WorkerTelemetry.RecordJob(job.JobType, "LeaseLost", + Stopwatch.GetElapsedTime(startedTimestamp).TotalMilliseconds); + } } diff --git a/Tiku.Infrastructure/Modules/JobsModule.cs b/Tiku.Infrastructure/Modules/JobsModule.cs index eecc9a6..ef193d6 100644 --- a/Tiku.Infrastructure/Modules/JobsModule.cs +++ b/Tiku.Infrastructure/Modules/JobsModule.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Tiku.Application.Jobs; using Tiku.Infrastructure.Jobs; @@ -8,6 +9,9 @@ internal static class JobsModule { internal static IServiceCollection AddJobsModule(this IServiceCollection services) { + services.TryAddSingleton(TimeProvider.System); + services.AddSingleton(BackgroundJobLeaseTiming.Default); + services.AddSingleton(); services.AddScoped(); services.AddScoped(provider => provider.GetRequiredService()); services.AddScoped(provider => provider.GetRequiredService()); diff --git a/Tiku.Infrastructure/Observability/WorkerTelemetry.cs b/Tiku.Infrastructure/Observability/WorkerTelemetry.cs index 21ae48b..d458e08 100644 --- a/Tiku.Infrastructure/Observability/WorkerTelemetry.cs +++ b/Tiku.Infrastructure/Observability/WorkerTelemetry.cs @@ -22,6 +22,9 @@ public static class WorkerTelemetry private static readonly Histogram IterationDuration = Meter.CreateHistogram("tiku.worker.iteration.duration", "ms"); + private static readonly Counter JobLeaseRenewalCounter = + Meter.CreateCounter("tiku.worker.job.lease_renewals"); + public static void RecordJob(string jobType, string status, double elapsedMilliseconds) { var tags = new TagList { { "job.type", jobType }, { "job.status", status } }; @@ -42,4 +45,9 @@ public static class WorkerTelemetry IterationCounter.Add(1, tags); IterationDuration.Record(elapsedMilliseconds, tags); } -} \ No newline at end of file + + public static void RecordJobLeaseRenewal(string outcome) + { + JobLeaseRenewalCounter.Add(1, new TagList { { "job.lease.outcome", outcome } }); + } +} diff --git a/Tiku.Infrastructure/Properties/AssemblyInfo.cs b/Tiku.Infrastructure/Properties/AssemblyInfo.cs index 185bc4b..1ad01bd 100644 --- a/Tiku.Infrastructure/Properties/AssemblyInfo.cs +++ b/Tiku.Infrastructure/Properties/AssemblyInfo.cs @@ -1,3 +1,4 @@ using System.Runtime.CompilerServices; [assembly: InternalsVisibleTo("Tiku.UnitTests")] +[assembly: InternalsVisibleTo("Tiku.IntegrationTests")] diff --git a/Tiku.IntegrationTests/Api/ApiTestFactory.cs b/Tiku.IntegrationTests/Api/ApiTestFactory.cs index abd70c4..3d71235 100644 --- a/Tiku.IntegrationTests/Api/ApiTestFactory.cs +++ b/Tiku.IntegrationTests/Api/ApiTestFactory.cs @@ -39,7 +39,8 @@ public sealed class ApiTestFactory( ISmsProvider? smsProvider = null, IAssetSecurityScanner? assetSecurityScanner = null, IReadOnlyDictionary? configurationOverrides = null, - DbCommandInterceptor? dbCommandInterceptor = null) : WebApplicationFactory + DbCommandInterceptor? dbCommandInterceptor = null, + Action? configureTestServices = null) : WebApplicationFactory { private readonly PostgresTestDatabase database = PostgresTestDatabase.Create(); @@ -127,6 +128,8 @@ public sealed class ApiTestFactory( services.RemoveAll(); services.AddSingleton(assetSecurityScanner); } + + configureTestServices?.Invoke(services); }); } @@ -513,4 +516,4 @@ public sealed class ApiTestFactory( base.Dispose(disposing); if (disposing) database.Dispose(); } -} \ No newline at end of file +} diff --git a/Tiku.IntegrationTests/Api/MonolithBackgroundProcessingTests.cs b/Tiku.IntegrationTests/Api/MonolithBackgroundProcessingTests.cs index f484605..497a366 100644 --- a/Tiku.IntegrationTests/Api/MonolithBackgroundProcessingTests.cs +++ b/Tiku.IntegrationTests/Api/MonolithBackgroundProcessingTests.cs @@ -2,10 +2,12 @@ using System.Net; using System.Text.Json; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Hosting; using Tiku.Application.Jobs; using Tiku.Domain.Operations; using Tiku.Domain.Tenancy; +using Tiku.Infrastructure.Jobs; using Tiku.Infrastructure.Persistence; namespace Tiku.IntegrationTests.Api; @@ -87,11 +89,117 @@ public sealed class MonolithBackgroundProcessingTests Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, pendingJobId)); } - private static async Task ProcessPendingAsync(ApiTestFactory factory, string workerId) + [Fact] + public async Task Active_job_lease_is_renewed_and_cannot_be_reclaimed() + { + var handler = new BlockingStatisticsHandler(); + await using var factory = CreateLeaseFactory(handler); + var jobId = await SeedJobAsync(factory, "Lease Renewal"); + + var firstWorker = ProcessPendingAsync(factory, "lease-renewal-a"); + await handler.Started.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await Task.Delay(TimeSpan.FromMilliseconds(700)); + + Assert.Equal(0, await ProcessPendingAsync(factory, "lease-renewal-b")); + handler.Release.TrySetResult(); + Assert.Equal(1, await firstWorker); + Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, jobId)); + } + + [Fact] + public async Task Lost_job_lease_cancels_old_worker_without_overwriting_new_owner() + { + var handler = new BlockingStatisticsHandler(); + await using var factory = CreateLeaseFactory(handler); + var jobId = await SeedJobAsync(factory, "Lease Ownership Loss"); + + var firstWorker = ProcessPendingAsync(factory, "lease-owner-a"); + await handler.Started.Task.WaitAsync(TimeSpan.FromSeconds(5)); + using (var scope = factory.CreateSystemScope("Transfer job lease ownership")) + { + var dbContext = scope.ServiceProvider.GetRequiredService(); + await dbContext.BackgroundJobs.Where(job => job.Id == jobId) + .ExecuteUpdateAsync(setters => setters + .SetProperty(job => job.LockedBy, "lease-owner-b") + .SetProperty(job => job.LockExpiresAt, DateTimeOffset.UtcNow.AddMinutes(1))); + } + + Assert.Equal(0, await firstWorker.WaitAsync(TimeSpan.FromSeconds(5))); + using var verificationScope = factory.CreateSystemScope("Verify transferred job lease ownership"); + var stored = await verificationScope.ServiceProvider.GetRequiredService() + .BackgroundJobs.AsNoTracking() + .SingleAsync(job => job.Id == jobId); + Assert.Equal(BackgroundJobStatus.Processing, stored.Status); + Assert.Equal("lease-owner-b", stored.LockedBy); + } + + [Fact] + public async Task Cancelled_worker_leaves_processing_lease_for_expiry_recovery() + { + var handler = new BlockingStatisticsHandler(); + await using var factory = CreateLeaseFactory(handler); + var jobId = await SeedJobAsync(factory, "Lease Cancellation Recovery"); + using var cancellation = new CancellationTokenSource(); + + var firstWorker = ProcessPendingAsync(factory, "lease-cancelled-a", cancellation.Token); + await handler.Started.Task.WaitAsync(TimeSpan.FromSeconds(5)); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(() => firstWorker); + + using (var scope = factory.CreateSystemScope("Verify cancelled worker lease")) + { + var stored = await scope.ServiceProvider.GetRequiredService() + .BackgroundJobs.AsNoTracking() + .SingleAsync(job => job.Id == jobId); + Assert.Equal(BackgroundJobStatus.Processing, stored.Status); + Assert.Equal("lease-cancelled-a", stored.LockedBy); + Assert.NotNull(stored.LockExpiresAt); + } + + await Task.Delay(TimeSpan.FromMilliseconds(700)); + Assert.Equal(1, await ProcessPendingAsync(factory, "lease-recovery-b")); + Assert.Equal(BackgroundJobStatus.Succeeded, await ReadStatusAsync(factory, jobId)); + } + + private static async Task ProcessPendingAsync( + ApiTestFactory factory, + string workerId, + CancellationToken cancellationToken = default) { using var scope = factory.CreateSystemScope($"Process jobs with {workerId}"); return await scope.ServiceProvider.GetRequiredService() - .ProcessPendingAsync(workerId, 1); + .ProcessPendingAsync(workerId, 1, cancellationToken: cancellationToken); + } + + private static ApiTestFactory CreateLeaseFactory(BlockingStatisticsHandler handler) + { + return new ApiTestFactory(configureTestServices: services => + { + services.RemoveAll(); + services.AddSingleton(handler); + services.RemoveAll(); + services.AddSingleton(new BackgroundJobLeaseTiming( + TimeSpan.FromMilliseconds(400), + TimeSpan.FromMilliseconds(50), + TimeSpan.FromMilliseconds(20))); + }); + } + + private static async Task SeedJobAsync(ApiTestFactory factory, string tenantName) + { + var tenantId = Guid.NewGuid(); + await factory.SeedAsync(new Tenant + { + Id = tenantId, + Slug = tenantId.ToString("N"), + Name = tenantName + }); + using var scope = factory.CreateSystemScope($"Seed {tenantName} job"); + return (await scope.ServiceProvider.GetRequiredService() + .EnqueueAsync(new CreateBackgroundJobCommand( + tenantId, + "statistics_aggregation", + JsonSerializer.SerializeToElement(new { scope = "tenant" })))).Id; } private static async Task ReadStatusAsync(ApiTestFactory factory, Guid jobId) @@ -103,4 +211,26 @@ public sealed class MonolithBackgroundProcessingTests .Select(job => job.Status) .SingleAsync(); } -} \ No newline at end of file + + private sealed class BlockingStatisticsHandler : IBackgroundJobHandler + { + private int calls; + + public string JobType => "statistics_aggregation"; + public TaskCompletionSource Started { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public async Task HandleAsync( + BackgroundJobExecutionContext context, + CancellationToken cancellationToken = default) + { + if (Interlocked.Increment(ref calls) == 1) + { + Started.TrySetResult(); + await Release.Task.WaitAsync(cancellationToken); + } + + return new BackgroundJobHandlerResult(JsonSerializer.SerializeToElement(new { completed = true })); + } + } +} diff --git a/Tiku.IntegrationTests/Infrastructure/DockerPostgresTestDefaults.cs b/Tiku.IntegrationTests/Infrastructure/DockerPostgresTestDefaults.cs new file mode 100644 index 0000000..c18db01 --- /dev/null +++ b/Tiku.IntegrationTests/Infrastructure/DockerPostgresTestDefaults.cs @@ -0,0 +1,28 @@ +using Npgsql; + +namespace Tiku.IntegrationTests.Infrastructure; + +internal static class DockerPostgresTestDefaults +{ + internal static string AdminConnectionString => CreateConnectionString("postgres", false); + + internal static string CreateConnectionString(string database, bool pooling = true) + { + return new NpgsqlConnectionStringBuilder + { + Host = "127.0.0.1", + Port = ParsePort(Environment.GetEnvironmentVariable("TIKU_POSTGRES_PORT")), + Database = database, + Username = Environment.GetEnvironmentVariable("TIKU_POSTGRES_USER") ?? "tiku", + Password = Environment.GetEnvironmentVariable("TIKU_POSTGRES_PASSWORD") ?? "tiku_dev", + Pooling = pooling, + Timeout = 5, + CommandTimeout = 60 + }.ConnectionString; + } + + private static int ParsePort(string? value) + { + return int.TryParse(value, out var port) && port is >= 1 and <= 65535 ? port : 5432; + } +} diff --git a/Tiku.IntegrationTests/Infrastructure/PostgresTestDatabase.cs b/Tiku.IntegrationTests/Infrastructure/PostgresTestDatabase.cs index 3415e4f..918cb49 100644 --- a/Tiku.IntegrationTests/Infrastructure/PostgresTestDatabase.cs +++ b/Tiku.IntegrationTests/Infrastructure/PostgresTestDatabase.cs @@ -119,7 +119,7 @@ internal sealed class PostgresTestDatabaseTemplate : IDisposable public static PostgresTestDatabaseTemplate Create() { var adminConnectionString = Environment.GetEnvironmentVariable("TIKU_TEST_POSTGRES_ADMIN") ?? - $"Host=localhost;Database=postgres;Username={Environment.UserName};Pooling=false;Timeout=5;Command Timeout=60"; + DockerPostgresTestDefaults.AdminConnectionString; var adminBuilder = new NpgsqlConnectionStringBuilder(adminConnectionString); if (!string.Equals(adminBuilder.Database, "postgres", StringComparison.OrdinalIgnoreCase)) throw new InvalidOperationException( @@ -171,4 +171,4 @@ internal sealed class PostgresTestDatabaseTemplate : IDisposable throw; } } -} \ No newline at end of file +} diff --git a/Tiku.IntegrationTests/ModulePersistenceBoundaryTests.cs b/Tiku.IntegrationTests/ModulePersistenceBoundaryTests.cs index 300b1a1..6a8c7a5 100644 --- a/Tiku.IntegrationTests/ModulePersistenceBoundaryTests.cs +++ b/Tiku.IntegrationTests/ModulePersistenceBoundaryTests.cs @@ -2,6 +2,7 @@ using Microsoft.Extensions.DependencyInjection; using Tiku.Application; using Tiku.Infrastructure; using Tiku.Infrastructure.Persistence; +using Tiku.IntegrationTests.Infrastructure; namespace Tiku.IntegrationTests; @@ -13,8 +14,7 @@ public sealed class ModulePersistenceBoundaryTests var services = new ServiceCollection(); services.AddLogging(); services.AddApplication(); - services.AddInfrastructure( - "Host=127.0.0.1;Port=5432;Database=tiku_module_boundary_test;Username=postgres;Password=unused"); + services.AddInfrastructure(DockerPostgresTestDefaults.CreateConnectionString("tiku_module_boundary_test")); using var provider = services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true }); using var scope = provider.CreateScope(); diff --git a/Tiku.IntegrationTests/PersistenceModelTests.cs b/Tiku.IntegrationTests/PersistenceModelTests.cs index 833b51c..a22dc10 100644 --- a/Tiku.IntegrationTests/PersistenceModelTests.cs +++ b/Tiku.IntegrationTests/PersistenceModelTests.cs @@ -13,6 +13,7 @@ using Tiku.Domain.Platform; using Tiku.Domain.QuestionBanks; using Tiku.Domain.Tenancy; using Tiku.Infrastructure.Persistence; +using Tiku.IntegrationTests.Infrastructure; namespace Tiku.IntegrationTests; @@ -20,7 +21,7 @@ public sealed class PersistenceModelTests { private static readonly DbContextOptions Options = new DbContextOptionsBuilder() - .UseNpgsql("Host=localhost;Database=tiku_model_tests;Username=postgres") + .UseNpgsql(DockerPostgresTestDefaults.CreateConnectionString("tiku_model_tests")) .Options; [Fact] @@ -932,4 +933,4 @@ public sealed class PersistenceModelTests ], keyProperties); } -} \ No newline at end of file +} diff --git a/Tiku.UnitTests/Tiku.UnitTests.csproj b/Tiku.UnitTests/Tiku.UnitTests.csproj index e5d0b14..f61c8eb 100644 --- a/Tiku.UnitTests/Tiku.UnitTests.csproj +++ b/Tiku.UnitTests/Tiku.UnitTests.csproj @@ -15,6 +15,7 @@ + diff --git a/Tiku.UnitTests/Worker/PeriodicWorkerExecutionModeTests.cs b/Tiku.UnitTests/Worker/PeriodicWorkerExecutionModeTests.cs new file mode 100644 index 0000000..4352c8b --- /dev/null +++ b/Tiku.UnitTests/Worker/PeriodicWorkerExecutionModeTests.cs @@ -0,0 +1,76 @@ +using Tiku.Worker; + +namespace Tiku.UnitTests.Worker; + +public sealed class PeriodicWorkerExecutionModeTests +{ + [Fact] + public async Task Multi_instance_mode_does_not_request_global_lock() + { + var processorLock = new RecordingProcessorLock(null); + + await using var lease = await PeriodicWorker.TryAcquireExecutionLeaseAsync( + PeriodicWorkerExecutionMode.MultiInstance, + processorLock, + "background-jobs", + CancellationToken.None); + + Assert.NotNull(lease); + Assert.Equal(0, processorLock.Attempts); + } + + [Fact] + public async Task Singleton_mode_skips_iteration_when_global_lock_is_unavailable() + { + var processorLock = new RecordingProcessorLock(null); + + var lease = await PeriodicWorker.TryAcquireExecutionLeaseAsync( + PeriodicWorkerExecutionMode.Singleton, + processorLock, + "commercial-billing", + CancellationToken.None); + + Assert.Null(lease); + Assert.Equal(1, processorLock.Attempts); + } + + [Fact] + public async Task Singleton_mode_returns_acquired_global_lock_lease() + { + var expected = new RecordingLease(); + var processorLock = new RecordingProcessorLock(expected); + + var lease = await PeriodicWorker.TryAcquireExecutionLeaseAsync( + PeriodicWorkerExecutionMode.Singleton, + processorLock, + "platform-approvals", + CancellationToken.None); + + Assert.Same(expected, lease); + Assert.Equal(1, processorLock.Attempts); + await lease!.DisposeAsync(); + Assert.True(expected.Disposed); + } + + private sealed class RecordingProcessorLock(IAsyncDisposable? lease) : IPeriodicProcessorLock + { + public int Attempts { get; private set; } + + public Task TryAcquireAsync(string processor, CancellationToken cancellationToken) + { + Attempts++; + return Task.FromResult(lease); + } + } + + private sealed class RecordingLease : IAsyncDisposable + { + public bool Disposed { get; private set; } + + public ValueTask DisposeAsync() + { + Disposed = true; + return ValueTask.CompletedTask; + } + } +} diff --git a/Tiku.Worker/Tiku.Worker.csproj b/Tiku.Worker/Tiku.Worker.csproj index eea2f78..6180482 100644 --- a/Tiku.Worker/Tiku.Worker.csproj +++ b/Tiku.Worker/Tiku.Worker.csproj @@ -12,6 +12,10 @@ + + + + diff --git a/Tiku.Worker/WorkerServices.cs b/Tiku.Worker/WorkerServices.cs index b94b76d..c537968 100644 --- a/Tiku.Worker/WorkerServices.cs +++ b/Tiku.Worker/WorkerServices.cs @@ -34,6 +34,12 @@ internal interface IPeriodicProcessorLock Task TryAcquireAsync(string processor, CancellationToken cancellationToken); } +internal enum PeriodicWorkerExecutionMode +{ + Singleton, + MultiInstance +} + internal sealed class WorkerStateReporter(NpgsqlDataSource dataSource) { private readonly DateTimeOffset startedAt = DateTimeOffset.UtcNow; @@ -127,7 +133,8 @@ internal abstract class PeriodicWorker( WorkerStateReporter stateReporter, string processorName, TimeSpan interval, - bool enabled) : BackgroundService + bool enabled, + PeriodicWorkerExecutionMode executionMode) : BackgroundService { protected abstract Task ProcessAsync(CancellationToken cancellationToken); @@ -138,7 +145,11 @@ internal abstract class PeriodicWorker( while (!stoppingToken.IsCancellationRequested) try { - await using var lease = await processorLock.TryAcquireAsync(processorName, stoppingToken); + await using var lease = await TryAcquireExecutionLeaseAsync( + executionMode, + processorLock, + processorName, + stoppingToken); if (lease is not null) { var iterationTimestamp = Stopwatch.GetTimestamp(); @@ -185,6 +196,27 @@ internal abstract class PeriodicWorker( { services.GetRequiredService().InitializeSystem(null, reason); } + + internal static Task TryAcquireExecutionLeaseAsync( + PeriodicWorkerExecutionMode mode, + IPeriodicProcessorLock processorLock, + string processorName, + CancellationToken cancellationToken) + { + return mode == PeriodicWorkerExecutionMode.Singleton + ? processorLock.TryAcquireAsync(processorName, cancellationToken) + : Task.FromResult(NoopAsyncDisposable.Instance); + } + + private sealed class NoopAsyncDisposable : IAsyncDisposable + { + internal static readonly NoopAsyncDisposable Instance = new(); + + public ValueTask DisposeAsync() + { + return ValueTask.CompletedTask; + } + } } internal sealed class TenantDomainWorker( @@ -195,7 +227,8 @@ internal sealed class TenantDomainWorker( IOptions workerOptions, ILogger logger) : PeriodicWorker(logger, processorLock, stateReporter, "tenant-domain-lifecycle", - TimeSpan.FromSeconds(Math.Clamp(domainOptions.Value.PollSeconds, 10, 3600)), workerOptions.Value.Enabled) + TimeSpan.FromSeconds(Math.Clamp(domainOptions.Value.PollSeconds, 10, 3600)), workerOptions.Value.Enabled, + PeriodicWorkerExecutionMode.Singleton) { protected override async Task ProcessAsync(CancellationToken cancellationToken) { @@ -213,7 +246,7 @@ internal sealed class SaasSubscriptionWorker( IOptions options, ILogger logger) : PeriodicWorker(logger, processorLock, stateReporter, "saas-subscription-lifecycle", TimeSpan.FromSeconds(60), - options.Value.Enabled) + options.Value.Enabled, PeriodicWorkerExecutionMode.Singleton) { protected override async Task ProcessAsync(CancellationToken cancellationToken) { @@ -232,7 +265,8 @@ internal sealed class FeatureUsageWorker( IOptions workerOptions, ILogger logger) : PeriodicWorker(logger, processorLock, stateReporter, "feature-usage-reconciliation", - TimeSpan.FromMinutes(Math.Clamp(featureOptions.Value.IntervalMinutes, 1, 1440)), workerOptions.Value.Enabled) + TimeSpan.FromMinutes(Math.Clamp(featureOptions.Value.IntervalMinutes, 1, 1440)), workerOptions.Value.Enabled, + PeriodicWorkerExecutionMode.Singleton) { protected override async Task ProcessAsync(CancellationToken cancellationToken) { @@ -250,7 +284,8 @@ internal sealed class BackgroundJobsWorker( IOptions options, ILogger logger) : PeriodicWorker(logger, processorLock, stateReporter, "background-jobs", - TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled) + TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled, + PeriodicWorkerExecutionMode.MultiInstance) { private readonly int batchSize = options.Value.JobBatchSize; private readonly int parallelism = options.Value.JobParallelism; @@ -279,7 +314,8 @@ internal sealed class AuthorizationCacheInvalidationWorker( IOptions options, ILogger logger) : PeriodicWorker(logger, processorLock, stateReporter, "authorization-cache-invalidations", - TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled) + TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled, + PeriodicWorkerExecutionMode.Singleton) { protected override async Task ProcessAsync(CancellationToken cancellationToken) { @@ -297,7 +333,7 @@ internal sealed class CommercialBillingWorker( IOptions options, ILogger logger) : PeriodicWorker(logger, processorLock, stateReporter, "commercial-billing", - TimeSpan.FromSeconds(60), options.Value.Enabled) + TimeSpan.FromSeconds(60), options.Value.Enabled, PeriodicWorkerExecutionMode.Singleton) { protected override async Task ProcessAsync(CancellationToken cancellationToken) { @@ -315,7 +351,8 @@ internal sealed class PlatformApprovalWorker( IOptions options, ILogger logger) : PeriodicWorker(logger, processorLock, stateReporter, "platform-approvals", - TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled) + TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled, + PeriodicWorkerExecutionMode.Singleton) { protected override async Task ProcessAsync(CancellationToken cancellationToken) {