fix: harden EfJobStore claim with atomic CAS and add Postgres integration test infra

chutch3 · chutch3 · commit f276198bc897 · 2026-06-07T13:26:04.000-04:00
diff --git a/api/API/JobRuntime/EfJobStore.cs b/api/API/JobRuntime/EfJobStore.cs
@@ -8,12 +8,6 @@ namespace API.JobRuntime;
 /// EF-backed <see cref="IJobStore"/>. Scoped (one per dispatcher tick). Claim selection reuses
 /// <see cref="JobReadySelection"/> so it matches the in-memory store the DF tests pin.
 /// </summary>
-/// <remarks>
-/// The claim loads the active/ready set and selects in memory. Under high concurrency two ticks could
-/// race the same row; hardening the claim with <c>SELECT … FOR UPDATE SKIP LOCKED</c> is deferred to the
-/// first real handler (step 2), where it is verified against Postgres. The runtime is not wired to real
-/// work yet, so no production claim race is exercised in step 1.
-/// </remarks>
 public class EfJobStore(JobsContext context) : IJobStore
 {
     public async Task<Job> EnqueueAsync(Job job, CancellationToken ct = default)
@@ -35,16 +29,45 @@ public async Task<Job> EnqueueAsync(Job job, CancellationToken ct = default)
     public async Task<Job?> ClaimNextReadyAsync(DateTime now, TimeSpan leaseDuration, int globalCap, int perResourceCap,
         CancellationToken ct = default)
     {
+        bool isRelational = context.Database.ProviderName != "Microsoft.EntityFrameworkCore.InMemory";
+
         List<Job> active = await context.JobQueue
             .Where(j => j.Status == JobStatus.Queued || j.Status == JobStatus.Running)
+            .AsNoTracking()
             .ToListAsync(ct);
 
-        if (JobReadySelection.PickClaimable(active, now, globalCap, perResourceCap) is not { } job)
+        if (JobReadySelection.PickClaimable(active, now, globalCap, perResourceCap) is not { } candidate)
             return null;
 
-        JobReadySelection.MarkClaimed(job, now, leaseDuration);
+        DateTime claimedUntil = now.Add(leaseDuration);
+
+        if (isRelational)
+        {
+            int updated = await context.JobQueue
+                .Where(j => j.Key == candidate.Key && j.Status == JobStatus.Queued)
+                .ExecuteUpdateAsync(s => s
+                    .SetProperty(j => j.Status, JobStatus.Running)
+                    .SetProperty(j => j.ScheduledFor, claimedUntil), ct);
+
+            if (updated == 0)
+                return null;
+
+            candidate.Status = JobStatus.Running;
+            candidate.ScheduledFor = claimedUntil;
+            return candidate;
+        }
+
+        // InMemory provider (tests): reload as tracked and mutate via SaveChanges.
+        Job? tracked = await context.JobQueue
+            .Where(j => j.Key == candidate.Key && j.Status == JobStatus.Queued)
+            .FirstOrDefaultAsync(ct);
+
+        if (tracked is null)
+            return null;
+
+        JobReadySelection.MarkClaimed(tracked, now, leaseDuration);
         await context.SaveChangesAsync(ct);
-        return job;
+        return tracked;
     }
 
     public async Task UpdateAsync(Job job, CancellationToken ct = default)
diff --git a/api/Tests/Integration/EfJobStoreConcurrencyTests.cs b/api/Tests/Integration/EfJobStoreConcurrencyTests.cs
@@ -0,0 +1,113 @@
+using API.JobRuntime.Interfaces;
+using API.Schema.JobsContext;
+using API.Schema.SeriesContext;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Npgsql;
+using Xunit;
+using JobEntity = API.Schema.JobsContext.Job;
+
+namespace API.Tests.Integration;
+
+[CollectionDefinition("postgres")]
+public class PostgresCollectionDefinition { }
+
+/// <summary>
+/// Verifies that two concurrent Dispatcher instances claiming from the same Postgres database
+/// each get at most one job — no double-execution.
+/// Requires postgres from docker-compose.test.yml (skips if not reachable).
+/// </summary>
+[Collection("postgres")]
+public class EfJobStoreConcurrencyTests : IAsyncLifetime
+{
+    private readonly PostgresFixture _pg = new();
+    private string? _dbName;
+    private KenkuApplicationFactory? _app1;
+    private KenkuApplicationFactory? _app2;
+
+    public async Task InitializeAsync()
+    {
+        if (!await PostgresReachableAsync())
+            return;
+        _dbName = await _pg.CreateDatabaseAsync();
+        string cs = _pg.GetConnectionString(_dbName);
+        var handler = new CountingHandler(() => Interlocked.Increment(ref _executions));
+        _app1 = new KenkuApplicationFactory { PostgresConnectionString = cs, ExtraJobHandlers = [handler] };
+        _app2 = new KenkuApplicationFactory { PostgresConnectionString = cs, ExtraJobHandlers = [handler] };
+    }
+
+    public async Task DisposeAsync()
+    {
+        _app1?.Dispose();
+        _app2?.Dispose();
+        if (_dbName is not null)
+            await _pg.DropDatabaseAsync(_dbName);
+    }
+
+    private int _executions;
+
+    [Fact]
+    public async Task ConcurrentDispatchers_ClaimSameJob_ExactlyOnce()
+    {
+        if (_app1 is null)
+            return; // Postgres not available — skip
+
+        using (var scope = _app1.Services.CreateScope())
+            await scope.ServiceProvider.GetRequiredService<IJobStore>().EnqueueAsync(
+                new JobEntity("CountingJob", "{}", DateTime.UtcNow));
+
+        var t1 = Task.Run(async () =>
+        {
+            using var scope = _app1.Services.CreateScope();
+            var dispatcher = scope.ServiceProvider.GetRequiredService<API.JobRuntime.Dispatcher>();
+            return await dispatcher.RunOnceAsync();
+        });
+        var t2 = Task.Run(async () =>
+        {
+            using var scope = _app2!.Services.CreateScope();
+            var dispatcher = scope.ServiceProvider.GetRequiredService<API.JobRuntime.Dispatcher>();
+            return await dispatcher.RunOnceAsync();
+        });
+
+        bool[] claimed = await Task.WhenAll(t1, t2);
+
+        Assert.Equal(1, claimed.Count(c => c));
+        Assert.Equal(1, _executions);
+    }
+
+    [Fact]
+    public async Task MigrationsApplied_AllContextsHaveSchema()
+    {
+        if (_app1 is null)
+            return; // Postgres not available — skip
+
+        using var scope = _app1.Services.CreateScope();
+        var sp = scope.ServiceProvider;
+
+        var jobsCtx = sp.GetRequiredService<global::API.Schema.JobsContext.JobsContext>();
+        Assert.Empty(await jobsCtx.Database.GetPendingMigrationsAsync());
+
+        var seriesCtx = sp.GetRequiredService<SeriesContext>();
+        Assert.Empty(await seriesCtx.Database.GetPendingMigrationsAsync());
+    }
+
+    private static async Task<bool> PostgresReachableAsync()
+    {
+        try
+        {
+            await using var conn = new NpgsqlConnection(PostgresFixture.AdminConnectionString);
+            await conn.OpenAsync(new CancellationTokenSource(TimeSpan.FromSeconds(3)).Token);
+            return true;
+        }
+        catch
+        {
+            return false;
+        }
+    }
+
+    private sealed class CountingHandler(Action onExecute) : IJobHandler
+    {
+        public string JobType => "CountingJob";
+        public Task ExecuteAsync(JobEntity job, CancellationToken ct) { onExecute(); return Task.CompletedTask; }
+    }
+}
diff --git a/api/Tests/Integration/KenkuApplicationFactory.cs b/api/Tests/Integration/KenkuApplicationFactory.cs
@@ -13,6 +13,7 @@
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Hosting;
 
 namespace API.Tests.Integration;
 
@@ -33,7 +34,12 @@ public sealed class KenkuApplicationFactory : WebApplicationFactory<Program>
         new ServiceCollection().AddEntityFrameworkInMemoryDatabase().BuildServiceProvider();
 
     /// <summary>Base URL of the local server outbound metadata requests should be redirected to.</summary>
-    public required string OutboundHttpTarget { get; init; }
+    public string OutboundHttpTarget { get; init; } = "http://localhost:1";
+
+    /// <summary>When set, the three core contexts (Series, Jobs, Actions) are backed by this Postgres
+    /// database instead of the default InMemory stores. Used by concurrency and migration tests that
+    /// need a real relational engine.</summary>
+    public string? PostgresConnectionString { get; init; }
 
     /// <summary>Optional stub for the connectors' HTTP edge, so a connector flow can be driven without
     /// real network access. When set, it replaces the registered <see cref="IHttpRequester"/>.</summary>
@@ -70,11 +76,22 @@ protected override void ConfigureWebHost(IWebHostBuilder builder)
         builder.UseSetting("Kenku:AppData", Path.Combine(Path.GetTempPath(), "kenku-test-" + _id));
         builder.ConfigureTestServices(services =>
         {
-            UseInMemory<SeriesContext>(services);
-            UseInMemory<NotificationsContext>(services);
-            UseInMemory<LibraryContext>(services);
-            UseInMemory<ActionsContext>(services);
-            UseInMemory<global::API.Schema.JobsContext.JobsContext>(services);
+            if (PostgresConnectionString is not null)
+            {
+                UseNpgsql<SeriesContext>(services, PostgresConnectionString);
+                UseNpgsql<global::API.Schema.JobsContext.JobsContext>(services, PostgresConnectionString);
+                UseNpgsql<ActionsContext>(services, PostgresConnectionString);
+                UseInMemory<NotificationsContext>(services);
+                UseInMemory<LibraryContext>(services);
+            }
+            else
+            {
+                UseInMemory<SeriesContext>(services);
+                UseInMemory<NotificationsContext>(services);
+                UseInMemory<LibraryContext>(services);
+                UseInMemory<ActionsContext>(services);
+                UseInMemory<global::API.Schema.JobsContext.JobsContext>(services);
+            }
 
             RouteOutboundHttp<MangaDexVolumeResolver>(services);
             RouteOutboundHttp<MangaDexSearchService>(services);
@@ -118,6 +135,20 @@ protected override void ConfigureWebHost(IWebHostBuilder builder)
         });
     }
 
+    protected override IHost CreateHost(IHostBuilder builder)
+    {
+        var host = base.CreateHost(builder);
+        if (PostgresConnectionString is not null)
+        {
+            using var scope = host.Services.CreateScope();
+            var sp = scope.ServiceProvider;
+            sp.GetRequiredService<SeriesContext>().Database.MigrateAsync().GetAwaiter().GetResult();
+            sp.GetRequiredService<global::API.Schema.JobsContext.JobsContext>().Database.MigrateAsync().GetAwaiter().GetResult();
+            sp.GetRequiredService<ActionsContext>().Database.MigrateAsync().GetAwaiter().GetResult();
+        }
+        return host;
+    }
+
     private void UseInMemory<TContext>(IServiceCollection services) where TContext : DbContext
     {
         services.RemoveAll<DbContextOptions<TContext>>();
@@ -127,6 +158,13 @@ private void UseInMemory<TContext>(IServiceCollection services) where TContext :
             .UseInternalServiceProvider(_efProvider));
     }
 
+    private static void UseNpgsql<TContext>(IServiceCollection services, string connectionString) where TContext : DbContext
+    {
+        services.RemoveAll<DbContextOptions<TContext>>();
+        services.RemoveAll<TContext>();
+        services.AddDbContext<TContext>(o => o.UseNpgsql(connectionString));
+    }
+
     // Replaces the typed client's primary handler so its (absolute) requests are redirected to the
     // local server. The resolver's own configuration (e.g. the User-Agent) is left intact.
     private void RouteOutboundHttp<TClient>(IServiceCollection services) where TClient : class =>
diff --git a/api/Tests/Integration/PostgresFixture.cs b/api/Tests/Integration/PostgresFixture.cs
@@ -0,0 +1,47 @@
+using Npgsql;
+
+namespace API.Tests.Integration;
+
+/// <summary>
+/// Manages per-test Postgres database lifecycle. Each test creates an isolated database
+/// (kenku_test_&lt;guid&gt;) and drops it in teardown. Requires postgres running via docker-compose.test.yml.
+/// Set TEST_POSTGRES_CONNECTION_STRING to override the default localhost:5433 endpoint.
+/// </summary>
+public sealed class PostgresFixture : IAsyncLifetime
+{
+    public static readonly string AdminConnectionString =
+        Environment.GetEnvironmentVariable("TEST_POSTGRES_CONNECTION_STRING")
+        ?? "Host=localhost;Port=5433;Username=kenku;Password=kenku_test;Database=postgres";
+
+    public Task InitializeAsync() => Task.CompletedTask;
+    public Task DisposeAsync() => Task.CompletedTask;
+
+    public async Task<string> CreateDatabaseAsync()
+    {
+        string name = $"kenku_test_{Guid.NewGuid():N}";
+        await using var conn = new NpgsqlConnection(AdminConnectionString);
+        await conn.OpenAsync();
+        await using var cmd = conn.CreateCommand();
+        cmd.CommandText = $"CREATE DATABASE \"{name}\"";
+        await cmd.ExecuteNonQueryAsync();
+        return name;
+    }
+
+    public async Task DropDatabaseAsync(string name)
+    {
+        await using var conn = new NpgsqlConnection(AdminConnectionString);
+        await conn.OpenAsync();
+        await using var kill = conn.CreateCommand();
+        kill.CommandText = $"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = '{name}'";
+        await kill.ExecuteNonQueryAsync();
+        await using var drop = conn.CreateCommand();
+        drop.CommandText = $"DROP DATABASE IF EXISTS \"{name}\"";
+        await drop.ExecuteNonQueryAsync();
+    }
+
+    public string GetConnectionString(string dbName)
+    {
+        var b = new NpgsqlConnectionStringBuilder(AdminConnectionString) { Database = dbName };
+        return b.ToString();
+    }
+}
diff --git a/api/Tests/Tests.csproj b/api/Tests/Tests.csproj
@@ -9,6 +9,7 @@
 
     <ItemGroup>
         <PackageReference Include="Microsoft.AspNetCore.Mvc.Testing" Version="10.0.8" />
+        <PackageReference Include="Npgsql" Version="10.0.1" />
         <PackageReference Include="Microsoft.EntityFrameworkCore.InMemory" Version="10.0.2" />
         <PackageReference Include="coverlet.collector" Version="6.0.4">
           <PrivateAssets>all</PrivateAssets>
diff --git a/api/docker-compose.test.yml b/api/docker-compose.test.yml
@@ -0,0 +1,14 @@
+services:
+  postgres:
+    image: postgres:15-alpine
+    environment:
+      POSTGRES_USER: kenku
+      POSTGRES_PASSWORD: kenku_test
+      POSTGRES_DB: postgres
+    ports:
+      - "5433:5432"
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U kenku"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
