From 7fdd173a61e1b3841fca3a812a6993713826db08 Mon Sep 17 00:00:00 2001 From: HeavenVR Date: Wed, 1 Jul 2026 11:25:33 +0200 Subject: [PATCH 1/4] test(email): cover the outbox claim query end-to-end against real Postgres Extract the delivery job's FOR UPDATE SKIP LOCKED claim query into EmailOutboxQueries.DueForDelivery as a single source of truth, and add an integration test that runs it through the same pooled OpenShockContext the Cron job uses. This exercises the real table/column names, the email_status enum mapping, LIMIT, the locking clause, and full entity materialization against a Testcontainers Postgres - so a schema rename or mapping regression fails the test rather than only surfacing in the Cron host at runtime. No behavior change to the job: identical SQL, server-side now(), same batch size. --- .../Tests/EmailOutboxPersistenceTests.cs | 23 +++++++++++++ Common/OpenShockDb/EmailOutboxQueries.cs | 32 +++++++++++++++++++ Cron/Jobs/EmailOutboxDeliveryJob.cs | 12 ++----- 3 files changed, 57 insertions(+), 10 deletions(-) create mode 100644 Common/OpenShockDb/EmailOutboxQueries.cs diff --git a/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs b/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs index 5755e190..e7289bfa 100644 --- a/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs +++ b/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs @@ -85,6 +85,29 @@ public async Task ClaimPredicate_SelectsDuePendingAndLapsedSending_NotTerminalOr await Assert.That(claimed.Count).IsEqualTo(2); } + [Test] + public async Task DueForDelivery_RunsTheJobsRawSql_AgainstTheRealSchema() + { + var factory = WebApplicationFactory.Services.GetRequiredService>(); + var recipient = TestHelper.UniqueEmail("outbox-claim-sql"); + + // Dated well into the past so it sorts first and the global LIMIT can never exclude it. + var id = await SeedAsync(factory, recipient, EmailStatus.Pending, DateTime.UtcNow - TimeSpan.FromDays(1)); + + // Resolve the pooled OpenShockContext exactly as the Cron delivery job does (DI-injected, not the + // factory) so this covers the same registration + enum-mapping path that runs in production. + using var scope = WebApplicationFactory.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + // Runs the delivery job's *actual* claim query (EmailOutboxQueries.DueForDelivery, the single source + // of truth the job uses) against real Postgres: the email_outbox table name, its column names, the + // email_status enum, LIMIT, FOR UPDATE SKIP LOCKED, and SELECT * -> full entity materialization + // (jsonb payload included). A schema rename breaks this test rather than only the Cron host at runtime. + var claimed = await db.EmailOutbox.DueForDelivery(EmailOutboxQueries.ClaimBatchSize).ToListAsync(); + + await Assert.That(claimed.Any(m => m.Id == id)).IsTrue(); + } + private static async Task SeedAsync(IDbContextFactory factory, string recipient, EmailStatus status, DateTime nextAttemptAt) { await using var db = await factory.CreateDbContextAsync(); diff --git a/Common/OpenShockDb/EmailOutboxQueries.cs b/Common/OpenShockDb/EmailOutboxQueries.cs new file mode 100644 index 00000000..f94cc104 --- /dev/null +++ b/Common/OpenShockDb/EmailOutboxQueries.cs @@ -0,0 +1,32 @@ +using Microsoft.EntityFrameworkCore; +using OpenShock.Common.Models; + +namespace OpenShock.Common.OpenShockDb; + +/// +/// Reusable queries over the table. Kept here (rather than inline in the +/// Cron delivery job) so the raw SQL is a single source of truth that integration tests can execute +/// directly - a table or column rename then breaks the test rather than surfacing only in production. +/// +public static class EmailOutboxQueries +{ + /// How many due rows the delivery job claims per batch. + public const int ClaimBatchSize = 50; + + /// + /// The delivery job's claim query: due rows - past their + /// next-attempt time, or whose lease has lapsed - oldest first, + /// capped at , locked with FOR UPDATE SKIP LOCKED so concurrent + /// runs take disjoint batches. Run it inside an explicit transaction to hold the locks for the claim. + /// + public static IQueryable DueForDelivery(this DbSet outbox, int batchSize) => + outbox.FromSql( + $""" + SELECT * FROM email_outbox + WHERE next_attempt_at <= now() + AND (status = {EmailStatus.Pending} OR status = {EmailStatus.Sending}) + ORDER BY next_attempt_at + LIMIT {batchSize} + FOR UPDATE SKIP LOCKED + """); +} diff --git a/Cron/Jobs/EmailOutboxDeliveryJob.cs b/Cron/Jobs/EmailOutboxDeliveryJob.cs index 59f45bef..c970be61 100644 --- a/Cron/Jobs/EmailOutboxDeliveryJob.cs +++ b/Cron/Jobs/EmailOutboxDeliveryJob.cs @@ -25,7 +25,7 @@ namespace OpenShock.Cron.Jobs; [CronJob("* * * * *")] // Every minute (https://crontab.guru/) public sealed class EmailOutboxDeliveryJob { - private const int BatchSize = 50; + private const int BatchSize = EmailOutboxQueries.ClaimBatchSize; private readonly OpenShockContext _db; private readonly IEmailOutboxDispatcher _dispatcher; @@ -88,15 +88,7 @@ private async Task> ClaimDueBatchAsync() await using var transaction = await _db.Database.BeginTransactionAsync(); - var due = await _db.EmailOutbox.FromSql( - $""" - SELECT * FROM email_outbox - WHERE next_attempt_at <= now() - AND (status = {EmailStatus.Pending} OR status = {EmailStatus.Sending}) - ORDER BY next_attempt_at - LIMIT {BatchSize} - FOR UPDATE SKIP LOCKED - """).ToListAsync(); + var due = await _db.EmailOutbox.DueForDelivery(BatchSize).ToListAsync(); if (due.Count == 0) { From f1ebee08bb1715daf4acec86ad7f4f45bf20a33f Mon Sep 17 00:00:00 2001 From: HeavenVR Date: Wed, 1 Jul 2026 12:33:23 +0200 Subject: [PATCH 2/4] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../Tests/EmailOutboxPersistenceTests.cs | 2 +- Common/OpenShockDb/EmailOutboxQueries.cs | 23 +++++++++++-------- 2 files changed, 14 insertions(+), 11 deletions(-) diff --git a/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs b/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs index e7289bfa..6a99db13 100644 --- a/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs +++ b/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs @@ -103,7 +103,7 @@ public async Task DueForDelivery_RunsTheJobsRawSql_AgainstTheRealSchema() // of truth the job uses) against real Postgres: the email_outbox table name, its column names, the // email_status enum, LIMIT, FOR UPDATE SKIP LOCKED, and SELECT * -> full entity materialization // (jsonb payload included). A schema rename breaks this test rather than only the Cron host at runtime. - var claimed = await db.EmailOutbox.DueForDelivery(EmailOutboxQueries.ClaimBatchSize).ToListAsync(); + var claimed = await db.EmailOutbox.DueForDelivery(1).ToListAsync(); await Assert.That(claimed.Any(m => m.Id == id)).IsTrue(); } diff --git a/Common/OpenShockDb/EmailOutboxQueries.cs b/Common/OpenShockDb/EmailOutboxQueries.cs index f94cc104..037fa042 100644 --- a/Common/OpenShockDb/EmailOutboxQueries.cs +++ b/Common/OpenShockDb/EmailOutboxQueries.cs @@ -19,14 +19,17 @@ public static class EmailOutboxQueries /// capped at , locked with FOR UPDATE SKIP LOCKED so concurrent /// runs take disjoint batches. Run it inside an explicit transaction to hold the locks for the claim. /// - public static IQueryable DueForDelivery(this DbSet outbox, int batchSize) => - outbox.FromSql( - $""" - SELECT * FROM email_outbox - WHERE next_attempt_at <= now() - AND (status = {EmailStatus.Pending} OR status = {EmailStatus.Sending}) - ORDER BY next_attempt_at - LIMIT {batchSize} - FOR UPDATE SKIP LOCKED - """); +public static IQueryable DueForDelivery(this DbSet outbox, int batchSize) +{ + if (batchSize <= 0) throw new ArgumentOutOfRangeException(nameof(batchSize)); + + return outbox.FromSql( + $""" + SELECT * FROM email_outbox + WHERE next_attempt_at <= now() + AND (status = {EmailStatus.Pending} OR status = {EmailStatus.Sending}) + ORDER BY next_attempt_at + LIMIT {batchSize} + FOR UPDATE SKIP LOCKED + """); } From ac8ebc19ab4b3d17dff5cf91cd96e4f699c9936b Mon Sep 17 00:00:00 2001 From: HeavenVR Date: Wed, 1 Jul 2026 12:38:11 +0200 Subject: [PATCH 3/4] fix(email): restore closing brace dropped by suggestion-apply in EmailOutboxQueries The 'Apply suggestions from code review' commit added the batchSize guard but dropped the EmailOutboxQueries class's closing brace, breaking the build (CS1513: } expected). Restore it and fix the method indentation. --- Common/OpenShockDb/EmailOutboxQueries.cs | 25 ++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/Common/OpenShockDb/EmailOutboxQueries.cs b/Common/OpenShockDb/EmailOutboxQueries.cs index 037fa042..ade6d04a 100644 --- a/Common/OpenShockDb/EmailOutboxQueries.cs +++ b/Common/OpenShockDb/EmailOutboxQueries.cs @@ -19,17 +19,18 @@ public static class EmailOutboxQueries /// capped at , locked with FOR UPDATE SKIP LOCKED so concurrent /// runs take disjoint batches. Run it inside an explicit transaction to hold the locks for the claim. /// -public static IQueryable DueForDelivery(this DbSet outbox, int batchSize) -{ - if (batchSize <= 0) throw new ArgumentOutOfRangeException(nameof(batchSize)); + public static IQueryable DueForDelivery(this DbSet outbox, int batchSize) + { + if (batchSize <= 0) throw new ArgumentOutOfRangeException(nameof(batchSize)); - return outbox.FromSql( - $""" - SELECT * FROM email_outbox - WHERE next_attempt_at <= now() - AND (status = {EmailStatus.Pending} OR status = {EmailStatus.Sending}) - ORDER BY next_attempt_at - LIMIT {batchSize} - FOR UPDATE SKIP LOCKED - """); + return outbox.FromSql( + $""" + SELECT * FROM email_outbox + WHERE next_attempt_at <= now() + AND (status = {EmailStatus.Pending} OR status = {EmailStatus.Sending}) + ORDER BY next_attempt_at + LIMIT {batchSize} + FOR UPDATE SKIP LOCKED + """); + } } From 70acaf95a599783a7b1207fdbea014636b520bd1 Mon Sep 17 00:00:00 2001 From: HeavenVR Date: Wed, 1 Jul 2026 12:58:04 +0200 Subject: [PATCH 4/4] test(email): move the outbox claim-query test to Cron.IntegrationTests DueForDelivery is a Cron delivery-job query, so its end-to-end test belongs beside the other Cron delivery tests, not in API.IntegrationTests. Move it into a new EmailOutboxQueryTests in Cron.IntegrationTests, where it seeds via the Cron host's DbContext factory and resolves the pooled OpenShockContext from the Cron host's own DI - an even closer match to the production claim path than the API host was. The EmailOutboxMessage round-trip/claim-predicate persistence tests stay in the API suite. No coverage lost. --- .../Tests/EmailOutboxPersistenceTests.cs | 23 --------- .../Tests/EmailOutboxQueryTests.cs | 47 +++++++++++++++++++ 2 files changed, 47 insertions(+), 23 deletions(-) create mode 100644 Cron.IntegrationTests/Tests/EmailOutboxQueryTests.cs diff --git a/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs b/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs index 6a99db13..5755e190 100644 --- a/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs +++ b/API.IntegrationTests/Tests/EmailOutboxPersistenceTests.cs @@ -85,29 +85,6 @@ public async Task ClaimPredicate_SelectsDuePendingAndLapsedSending_NotTerminalOr await Assert.That(claimed.Count).IsEqualTo(2); } - [Test] - public async Task DueForDelivery_RunsTheJobsRawSql_AgainstTheRealSchema() - { - var factory = WebApplicationFactory.Services.GetRequiredService>(); - var recipient = TestHelper.UniqueEmail("outbox-claim-sql"); - - // Dated well into the past so it sorts first and the global LIMIT can never exclude it. - var id = await SeedAsync(factory, recipient, EmailStatus.Pending, DateTime.UtcNow - TimeSpan.FromDays(1)); - - // Resolve the pooled OpenShockContext exactly as the Cron delivery job does (DI-injected, not the - // factory) so this covers the same registration + enum-mapping path that runs in production. - using var scope = WebApplicationFactory.Services.CreateScope(); - var db = scope.ServiceProvider.GetRequiredService(); - - // Runs the delivery job's *actual* claim query (EmailOutboxQueries.DueForDelivery, the single source - // of truth the job uses) against real Postgres: the email_outbox table name, its column names, the - // email_status enum, LIMIT, FOR UPDATE SKIP LOCKED, and SELECT * -> full entity materialization - // (jsonb payload included). A schema rename breaks this test rather than only the Cron host at runtime. - var claimed = await db.EmailOutbox.DueForDelivery(1).ToListAsync(); - - await Assert.That(claimed.Any(m => m.Id == id)).IsTrue(); - } - private static async Task SeedAsync(IDbContextFactory factory, string recipient, EmailStatus status, DateTime nextAttemptAt) { await using var db = await factory.CreateDbContextAsync(); diff --git a/Cron.IntegrationTests/Tests/EmailOutboxQueryTests.cs b/Cron.IntegrationTests/Tests/EmailOutboxQueryTests.cs new file mode 100644 index 00000000..d69df67d --- /dev/null +++ b/Cron.IntegrationTests/Tests/EmailOutboxQueryTests.cs @@ -0,0 +1,47 @@ +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using OpenShock.Common.Models; +using OpenShock.Common.OpenShockDb; + +namespace OpenShock.Cron.IntegrationTests.Tests; + +/// +/// Runs the delivery job's actual claim query (, the single +/// source of truth the job uses) against real Postgres via the Cron host's own pooled context - the same +/// registration + enum-mapping path that runs in production. Guards the raw SQL (table/column names, the +/// email_status enum, LIMIT, FOR UPDATE SKIP LOCKED, SELECT * materialization) so a schema rename breaks +/// this test rather than only surfacing in the Cron host at runtime. +/// +public sealed class EmailOutboxQueryTests +{ + [ClassDataSource(Shared = SharedType.PerTestSession)] + public required CronApplicationFactory Factory { get; init; } + + [Test] + public async Task DueForDelivery_RunsTheJobsRawSql_AgainstTheRealSchema() + { + var recipient = $"outbox-claim-sql-{Guid.CreateVersion7():N}@test.org"; + + Guid id; + await using (var db = await Factory.DbContextFactory.CreateDbContextAsync()) + { + var message = EmailOutboxMessage.Create( + EmailType.PasswordReset, recipient, "Claim", + new Dictionary { ["k"] = "v" }); + // Dated well into the past so it sorts first and the global LIMIT can never exclude it. + message.NextAttemptAt = DateTime.UtcNow - TimeSpan.FromDays(1); + db.EmailOutbox.Add(message); + await db.SaveChangesAsync(); + id = message.Id; + } + + // Resolve the pooled OpenShockContext exactly as the Cron delivery job does (DI-injected) so this + // covers the same registration + enum-mapping path that runs in production. + using var scope = Factory.Services.CreateScope(); + var db2 = scope.ServiceProvider.GetRequiredService(); + + var claimed = await db2.EmailOutbox.DueForDelivery(1).ToListAsync(); + + await Assert.That(claimed.Any(m => m.Id == id)).IsTrue(); + } +}