fix: serialise Postgres claims with FOR UPDATE SKIP LOCKED and seed reconciler test data before factory start

Author: chutch3

api/API/JobRuntime/EfJobStore.cs

@@ -29,45 +29,51 @@ 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";
+        if (context.Database.ProviderName != "Microsoft.EntityFrameworkCore.InMemory")
+            return await ClaimRelationalAsync(now, leaseDuration, globalCap, perResourceCap, ct);
 
         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 { } candidate)
             return null;
 
-        DateTime claimedUntil = now.Add(leaseDuration);
+        JobReadySelection.MarkClaimed(candidate, now, leaseDuration);
+        await context.SaveChangesAsync(ct);
+        return candidate;
+    }
 
-        if (isRelational)
+    private async Task<Job?> ClaimRelationalAsync(DateTime now, TimeSpan leaseDuration, int globalCap, int perResourceCap,
+        CancellationToken ct)
+    {
+        await using var txn = await context.Database.BeginTransactionAsync(ct);
+        try
         {
-            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)
+            List<Job> active = await context.JobQueue
+                .FromSqlRaw("""
+                    SELECT * FROM "JobQueue"
+                    WHERE "Status" IN (0, 1)
+                    FOR UPDATE SKIP LOCKED
+                    """)
+                .ToListAsync(ct);
+
+            if (JobReadySelection.PickClaimable(active, now, globalCap, perResourceCap) is not { } candidate)
+            {
+                await txn.RollbackAsync(ct);
                 return null;
+            }
 
-            candidate.Status = JobStatus.Running;
-            candidate.ScheduledFor = claimedUntil;
+            JobReadySelection.MarkClaimed(candidate, now, leaseDuration);
+            await context.SaveChangesAsync(ct);
+            await txn.CommitAsync(ct);
             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 tracked;
+        catch
+        {
+            await txn.RollbackAsync(ct);
+            throw;
+        }
     }
 
     public async Task UpdateAsync(Job job, CancellationToken ct = default)

api/Tests/Integration/ReconcilerLoopTests.cs

@@ -1,4 +1,5 @@
 using API.JobRuntime.Handlers;
+using API.Schema.ActionsContext;
 using API.Schema.SeriesContext;
 using Microsoft.EntityFrameworkCore;
 using Xunit;
@@ -22,13 +23,40 @@ public async Task InitializeAsync()
     {
         if (!await _postgres.IsReachableAsync()) return;
         _dbName = await _postgres.CreateDatabaseAsync();
+        string cs = _postgres.GetConnectionString(_dbName);
+        Directory.CreateDirectory(_libDir);
+
+        await MigrateAsync<SeriesContext>(cs);
+        await MigrateAsync<global::API.Schema.JobsContext.JobsContext>(cs);
+        await MigrateAsync<ActionsContext>(cs);
+
+        var opts = new DbContextOptionsBuilder<SeriesContext>().UseNpgsql(cs).Options;
+        await using (var ctx = new SeriesContext(opts))
+        {
+            var library = new FileLibrary(_libDir, "Lib");
+            ctx.FileLibraries.Add(library);
+            var manga = new Series("Test", "", "http://x/c.jpg", SeriesReleaseStatus.Continuing, [], [], [], [], library);
+            ctx.Series.Add(manga);
+            var chapter = new Chapter(manga, "1", null, null);
+            ctx.Chapters.Add(chapter);
+            var sourceId = new SourceId<Chapter>(chapter, "StubConnector", "site-id-1", "http://stub.test/1", true);
+            ctx.MangaConnectorToChapter.Add(sourceId);
+            await ctx.SaveChangesAsync();
+        }
+
         _app = new KenkuApplicationFactory
         {
-            PostgresConnectionString = _postgres.GetConnectionString(_dbName),
+            PostgresConnectionString = cs,
             RunStartup = true,
             DispatcherCaps = (0, 0),
         };
-        Directory.CreateDirectory(_libDir);
+    }
+
+    private static async Task MigrateAsync<TContext>(string cs) where TContext : DbContext
+    {
+        var opts = new DbContextOptionsBuilder<TContext>().UseNpgsql(cs).Options;
+        await using var ctx = (TContext)Activator.CreateInstance(typeof(TContext), opts)!;
+        await ctx.Database.MigrateAsync();
     }
 
     public async Task DisposeAsync()
@@ -44,20 +72,6 @@ public async Task DownloadReconciler_OnFirstTick_EnqueuesDownloadJobForRequested
     {
         if (_app is null) return; // skip: Postgres not available
 
-        await _app.WithSeriesContext(async ctx =>
-        {
-            var library = new FileLibrary(_libDir, "Lib");
-            ctx.FileLibraries.Add(library);
-            var manga = new Series("Test", "", "http://x/c.jpg", SeriesReleaseStatus.Continuing, [], [], [], [], library);
-            ctx.Series.Add(manga);
-            var chapter = new Chapter(manga, "1", null, null);
-            ctx.Chapters.Add(chapter);
-            var sourceId = new SourceId<Chapter>(chapter, "StubConnector", "site-id-1", "http://stub.test/1", true);
-            ctx.MangaConnectorToChapter.Add(sourceId);
-            await ctx.SaveChangesAsync();
-            return 0;
-        });
-
         bool appeared = false;
         var deadline = DateTime.UtcNow.AddSeconds(10);
         while (DateTime.UtcNow < deadline)