fix(events): delete expired inbox/outbox rows with FOR UPDATE SKIP LOCKED

[yilmaztayfun]

Why

With multiple replicas (seen in test on vnext-morph-idm-worker-inbox: ~1000 errors/week, on every pod), inbox cleanup logs Error cleaning up old inbox messages with:

AetherDbConcurrencyException: The database operation was expected to affect 1 row(s), but actually affected 0 row(s)

Each worker's cleanup read the same oldest Processed rows (ORDER BY … LIMIT) without a lock and removed them via RemoveRange. The first commit won; every other worker's per-row DELETE WHERE "Id" = … hit 0 rows and EF raised a concurrency exception. No data was lost, but it produced Error-level noise on every pod. It happened constantly because CleanupInterval was documented but never read, so cleanup ran on every polling cycle (every 100 ms when busy).

The outbox had the same pattern, and it was worse there. The cleanup exception wasn't caught locally, so RunAsync returned 0 even after publishing. OutboxBackgroundService then backed off to the idle interval, which adds publish latency under load.

What changed

• New IInboxCleanupStore / IOutboxCleanupStore (Abstractions)
  • Npgsql: NpgsqlInboxCleanupStore / NpgsqlOutboxCleanupStore run a single DELETE … WHERE "Id" IN (SELECT … ORDER BY … LIMIT … FOR UPDATE SKIP LOCKED). Workers delete disjoint batches, never wait on each other and never fail on each other's rows. Same statement shape as the existing Npgsql lease stores.
  • EF Core fallback (SQL Server and others): EfCore*CleanupStore reads ids and then runs ExecuteDeleteAsync. A row another worker already deleted is simply not counted, with no exception.
  • Registration: follows the lease-store model. TryAddScoped fallback in AddAetherInbox/AddAetherOutbox, AddScoped override in AddAetherNpgsql, so registration order doesn't matter.
• CleanupInterval is now honoured (inbox), and CleanupInterval + CleanupBatchSize are added to AetherOutboxOptions.
  • A full batch keeps cleanup due on the next cycle, so a backlog still drains one batch per cycle.
  • A short batch or a failure defers cleanup by the interval.
  • Timing uses the monotonic tick count and is due on the first cycle after startup.
• Outbox cleanup has its own try/catch and can no longer discard the publish count.
• IInboxStore.CleanupOldMessagesAsync is [Obsolete] and no longer called by Aether. EfCoreInboxStore's implementation delegates to the set-based delete.
• Docs: docs/inbox-outbox/INTERNALS.md (new Retention Cleanup section) and README.md.

Behaviour changes for consumers

• Cleanup now runs at most once per hour per worker by default, unless a backlog is draining. Set CleanupInterval = TimeSpan.Zero to get the old every-cycle behaviour.
• Outbox cleanup batch size is now CleanupBatchSize (default 1000) instead of BatchSize (100).
• Custom overrides of IInboxStore.CleanupOldMessagesAsync are no longer invoked. Register an IInboxCleanupStore or override InboxProcessor.CleanupOldMessagesAsync instead.

Tests

• NpgsqlCleanupStoreTests (Testcontainers, 7 tests):
  • Only expired Processed rows are deleted, oldest first.
  • Worker A holds its batch's locks in an open transaction while worker B deletes the remaining rows without blocking (10 s guard) and without an exception.
  • The EF fallback returns 0 for rows that are already gone.
  • The Npgsql stores win in both registration orders.
  • Mutation check: removing SKIP LOCKED makes the concurrency test time out.
• CleanupScheduleTests (6 tests): interval / full-batch / failure deferral for both processors, and outbox cleanup failure no longer zeroes RunAsync's count. It shares an xUnit collection with OutboxProcessorTraceTests, because that test's ActivityListener is process-wide and would otherwise capture Outbox.Process spans from these tests running in parallel.
• Results: Infrastructure.Tests 230/230 (3 runs), AspNetCore.Tests 15/15, solution builds with 0 errors.
• Postgres.Tests: 142/143. The one failure, MultiSchemaUnitOfWorkTests.Exceeds_max_context_limit_throws, also fails on clean master and is unrelated to this change.

Reviewer notes

• Not exercised against a real SQL Server. The SQL Server provider doesn't register lease stores yet, so inbox/outbox isn't functional there anyway, and the EF fallback is provider-agnostic LINQ.
• No migration is needed. The existing IX_InboxMessages_Cleanup(Status, HandledTime) and IX_OutboxMessages_Cleanup(ProcessedAt, CreatedAt) indexes cover the new queries.

🤖 Generated with Claude Code

[coderabbitai]

Important

• 🔍 Trigger review

This repository does not receive automatic reviews because it has fewer than 10 stars.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Advanced

Run ID: 0b94f50e-c4ca-4d2f-bfa5-8ac8da80e8f5

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[sonarqubecloud]

Quality Gate failed

Failed conditions
0.0% Coverage on New Code (required ≥ 80%)
C Security Rating on New Code (required ≥ A)

See analysis details on SonarQube Cloud

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE