A set of building blocks for reliable database-driven messaging and idempotent processing:

Outbox, Inbox or OnceExecutor? Outbox solves the outgoing side: publish a message atomically with the database change that caused it. Inbox and OnceExecutor both solve the incoming side, but differently:

• Dex.Cap.OnceExecutor runs your logic inline, inside the consumer, and remembers the idempotency key so a redelivery short-circuits. The message body is never stored. Use it when the handler is fast and you are happy to hold the message on the broker while it runs.
• Dex.Cap.Inbox stores the message, lets you acknowledge the source immediately, and processes it later in a background worker with its own retries, dead-lettering and observability. Use it when processing is slow, when the source must not wait (an HTTP request), or when you want the retry policy to be yours rather than the broker's. Deduplication is built in and is not optional.

Implementation of the Outbox pattern, which allows you to atomically insert outgoing commands for asynchronous execution into the database together with the main operation. The template guarantees that commands will not be lost.

• Performs the operation and stores messages in the outbox queue inside the same transaction.
• All messages of a logical operation are linked by a single CorrelationId.
• Avoids closures — state is passed explicitly.
• Allows registering services required by the operation.
• Verifies the success of the operation (verifySucceeded callback).
• Discriminates messages by type via OutboxTypeId for serialization and dispatching.

// Define an outbox message
public class OrderCreatedOutboxMessage : IOutboxMessage
{
    public static string OutboxTypeId => "3961131e-3961-4c38-8a30-09b91cb56d60";

    // Optional, default = true.
    // If false — the message will NOT be auto-published by PublisherOutboxHandler,
    // an explicit IOutboxMessageHandler<T> must be registered.
    public static bool AllowAutoPublishing => true;

    // Optional, default = false.
    // If true — the row is deleted from the DB immediately after the handler completes
    // (recommended for large or very frequent messages to keep the table small).
    public static bool DeleteImmediately => false;

    public string Args { get; init; } = "";
}

// Optional: implement a dedicated handler. If none is registered and AllowAutoPublishing == true,
// PublisherOutboxHandler<T> (from Dex.Cap.Outbox.OnceExecutor.MassTransit) will publish the message.
public class OrderCreatedOutboxMessageHandler(ILogger<OrderCreatedOutboxMessageHandler> logger)
    : IOutboxMessageHandler<OrderCreatedOutboxMessage>
{
    public Task Process(OrderCreatedOutboxMessage message, CancellationToken cancellationToken)
    {
        logger.LogInformation("Processed at {Now}, Args: {Args}", DateTime.UtcNow, message.Args);
        return Task.CompletedTask;
    }
}

// ConfigureServices
using Dex.Cap.Outbox.Ef.Extensions;
using Dex.Cap.Outbox.OnceExecutor.MassTransit.Extensions; // for AddOutboxPublisher()

services.AddOutbox<AppDbContext>();                         // core services + EF data provider
services.AddDefaultOutboxScheduler<AppDbContext>(           // background processing + cleanup
    periodSeconds: 30,                                       // poll interval
    cleanupDays: 30);                                        // delete processed messages older than N days

services.AddScoped<IOutboxMessageHandler<OrderCreatedOutboxMessage>,
                   OrderCreatedOutboxMessageHandler>();      // dedicated handler (optional)

services.AddOutboxPublisher();                              // auto-publish any IOutboxMessage
                                                            // with AllowAutoPublishing == true

// OnModelCreating
modelBuilder.OutboxModelCreating();

// Remark: when using optimistic concurrency in PostgreSQL — exclude OutboxEnvelope:
modelBuilder.UseXminAsConcurrencyToken(ignoreTypes: typeof(OutboxEnvelope));

public class OrderService(IOutboxService outbox, AppDbContext db)
{
    public async Task PlaceOrderAsync(string name, CancellationToken ct)
    {
        await db.Users.AddAsync(new User { Name = name }, ct);
        await outbox.EnqueueAsync(new OrderCreatedOutboxMessage { Args = name }, cancellationToken: ct);

        await db.SaveChangesAsync(ct); // atomic: domain change + outbox row
    }
}

For automatic retries on transient DB errors and transactional guarantees, wrap the call in db.ExecuteInTransactionAsync(...) from Dex.Cap.Common.Ef (see below).

Outbox messages must implement Dex.Cap.Common.Interfaces.IOutboxMessage.

Configurable via IOptions<OutboxOptions> (defaults shown):

By default a no-op strategy is used (retry at UtcNow). To space retries out:

using Dex.Cap.Outbox.Extensions;

services.AddOutbox<AppDbContext>((_, configurator) =>
{
    configurator.UseOutboxIncrementalRetryStrategy(TimeSpan.FromSeconds(30));
});

Or provide a custom IOutboxRetryStrategy:

services.AddOutbox<AppDbContext>((_, configurator) =>
{
    configurator.RetryStrategy = new MyExponentialBackoffStrategy();
});

IOutboxTypeDiscriminatorProvider scans the current AppDomain at first access and:

• builds the map of OutboxTypeId → CLR type;
• reports which discriminators the current process can handle (i.e. has a registered handler);
• collects discriminators marked with DeleteImmediately == true.

This is what enables a single outbox table to be safely shared between multiple services that consume different subsets of message types.

AddDefaultOutboxScheduler registers a health check named outbox-scheduler (tagged outbox-scheduler) that surfaces background-service failures via ASP.NET HealthChecks.

Bridges the outbox to MassTransit.

Open-generic handler that automatically publishes any IOutboxMessage (with AllowAutoPublishing == true) to MassTransit. Use it when the outbox row itself is the integration event.

services.AddOutbox<AppDbContext>();
services.AddOutboxPublisher(); // registers IOutboxMessageHandler<> -> PublisherOutboxHandler<>

Base classes for writing custom outbox handlers that need:

• Idempotency — guarantees a single execution per IIdempotentKey.IdempotentKey (or MassTransit MessageId if IIdempotentKey is not implemented); duplicates exit silently.
• Transactionality — wraps the handler in ExecuteInTransactionAsync.

Requires the RabbitMQ plugin rabbitmq_delayed_message_exchange.

Base consumers for messages coming directly from MassTransit (not via the outbox). Handle errors, write logs, support Defer (throw DeferConsumerException → republish into delay_exchange after the specified interval).

IdempotentConsumer adds idempotency on top of the base consumer.

Implementation of the Transactional Inbox pattern: an incoming message is stored first, acknowledged to its source immediately, and processed later by a background worker inside a single database transaction.

• Deduplicates by (MessageId, ConsumerId) — a redelivery never produces a second row.
• Decouples the source from processing: the broker (or the HTTP caller) is released as soon as the row is committed.
• Processes the handler and the status change in one transaction: the business effect and "message handled" either both commit or both roll back.
• Retries with a configurable strategy, then dead-letters — a failing message never disappears silently and never loops forever.
• Scales horizontally: workers claim messages with FOR UPDATE SKIP LOCKED plus a lease, so concurrent instances never take the same message.
• Transport-agnostic: the core knows nothing about MassTransit, HTTP or gRPC.

The transaction covers your database only. Any external call made by the handler (HTTP, broker, push service) will be repeated if the message is retried, so those calls must be idempotent on their own — via a provider-side idempotency key or a lookup before the call. The inbox gives you effectively once, not exactly once.

// Define an inbox message
public class OrderCreatedInboxCommand : IInboxMessage
{
    // Discriminator. It is written to the database and read back after a restart or a deploy,
    // so it must never change for an existing type.
    public static string InboxTypeId => "3961131e-3961-4c38-8a30-09b91cb56d60";

    public string Args { get; init; } = "";
}

// A handler is mandatory: only messages with a registered handler are fetched by this service.
// It runs inside the processing transaction, so it must NOT commit by itself.
public class OrderCreatedInboxCommandHandler(AppDbContext db) : IInboxMessageHandler<OrderCreatedInboxCommand>
{
    public async Task Process(OrderCreatedInboxCommand message, CancellationToken cancellationToken)
    {
        await db.Orders.AddAsync(new Order { Args = message.Args }, cancellationToken);
        // no SaveChanges here — the inbox commits the effect together with the message status
    }
}

// ConfigureServices
using Dex.Cap.Inbox.Ef.Extensions;

services.AddInbox<AppDbContext>();                        // core services + EF data provider
services.AddDefaultInboxScheduler<AppDbContext>(          // background processing + cleanup
    periodSeconds: 30,                                     // pause only when the queue is drained
    cleanupDays: 30);                                      // retention == deduplication window

services.AddScoped<IInboxMessageHandler<OrderCreatedInboxCommand>, OrderCreatedInboxCommandHandler>();

// OnModelCreating
modelBuilder.InboxModelCreating();

// Remark: when using optimistic concurrency in PostgreSQL — exclude InboxEnvelope:
modelBuilder.UseXminAsConcurrencyToken(ignoreTypes: typeof(InboxEnvelope));

public class OrderCreatedConsumer(IInboxService inbox) : IConsumer<OrderCreatedEvent>
{
    // Stable per consumer: it is part of the deduplication key.
    private const string ConsumerId = nameof(OrderCreatedConsumer);

    public async Task Consume(ConsumeContext<OrderCreatedEvent> context)
    {
        var messageId = context.MessageId ?? throw new InvalidOperationException("MessageId is required for deduplication");

        // A duplicate is a normal outcome of at-least-once delivery, not an error:
        // returning normally acknowledges the message instead of sending it to the error queue.
        await inbox.EnqueueAsync(
            new OrderCreatedInboxCommand { Args = context.Message.Args },
            new InboxMessageIdentity(messageId.ToString("N"), ConsumerId),
            cancellationToken: context.CancellationToken);
    }
}

[HttpPost("orders")]
public async Task<IActionResult> CreateOrder(
    [FromHeader(Name = "Idempotency-Key")] string idempotencyKey,
    CreateOrderRequest request,
    CancellationToken cancellationToken)
{
    var status = await inbox.EnqueueAsync(
        new OrderCreatedInboxCommand { Args = request.Args },
        new InboxMessageIdentity(idempotencyKey, "POST /orders"),
        cancellationToken: cancellationToken);

    // Both outcomes are success for the caller: the message is accepted exactly once either way.
    return status == InboxEnqueueStatus.Accepted ? Accepted() : Ok();
}

Unlike IOutboxService.EnqueueAsync, this method persists immediately in its own transaction: the point of the inbox is to store the message before the source is acknowledged, and there is no business work to be atomic with. For the same reason enqueuing inside a transaction of your own is rejected with an InboxException: an enclosing DbContext transaction or an ambient TransactionScope would roll the message back while the source has already been acknowledged.

InboxOptions (validated on host start):

InboxHandlerOptions (scheduler): Period 30s, CleanupInterval 1h, CleanupOlderThan 30d, HandlerInitDelay 5–15s, CleanerInitDelay 20–40s.

Period is a pause between drained cycles, not a throughput limit: while the worker keeps claiming full batches it continues without pausing.

                    ┌──────────────────────────── retry (attempts left) ────────────────────────────┐
                    │                                                                               │
   Enqueue ──▶ [New] ──▶ claim (FOR UPDATE SKIP LOCKED + lease) ──▶ handler ──┬── success ──▶ [Succeeded] ──▶ cleanup after CleanupOlderThan
                    │                                                          │
                    └──────────────────────────────────────────────────────────┴── failure ──▶ [Failed] ──▶ … ──▶ [DeadLettered] (terminal, kept for review)

There is no separate "in progress" state on purpose: the in-flight marker is the lease (LockId + LockExpirationTimeUtc). A crashed worker lets the lease expire and the message returns to the queue, whereas a status flag would stay stuck forever.

CleanupOlderThan deletes processed messages, and the row is the deduplication key. Once it is gone, a redelivery of that message is accepted as new. Keep the retention comfortably above the maximum redelivery horizon of the source. Cleanup only removes messages of the discriminators this service handles, so a co-tenant sharing the table keeps its own deduplication window regardless of your CleanupOlderThan.

DeadLettered messages are never deleted by cleanup — they exist to be investigated, and DeadLetteredJobCount reports how many of this service's own messages are waiting.

Once you have fixed the cause, return the message to processing through IInboxDeadLetterService, not by editing the table by hand. Returning a message takes a coordinated reset of several columns at once — status, attempt count, lease and the fetch index — and getting one of them wrong buries the message again on the next cycle, so the reset is encapsulated:

// one message, by its (MessageId, ConsumerId):
await deadLetterService.RequeueAsync(new InboxMessageIdentity("message-id", "consumer-id"));

// or every dead lettered message of this service at once, after fixing a systemic cause:
var returned = await deadLetterService.RequeueAllAsync();

Both operations touch only this service's own discriminators and only rows that are actually dead lettered, so a shared table is safe. DeadLetteredJobCount drops on its own as the messages return to processing.

services.AddInbox<AppDbContext>(
    options => options.Retries = 6,
    (_, configurator) =>
        configurator.UseExponentialStrategy(baseDelay: TimeSpan.FromMinutes(1), maxDelay: TimeSpan.FromMinutes(30)));

The delay is measured from the moment the attempt failed, so a backlog does not eat it.

Default is no extra delay — the next cycle picks the message up. UseIncrementalStrategy(interval) and UseExponentialStrategy(baseDelay, maxDelay) are also available.

Two things to size correctly, or the strategy is decorative:

• A delay below Period (default 30s) is not observable: the next attempt happens on the next cycle anyway.
• The exponent is bounded by Retries. The attempt that exhausts the limit dead-letters the message without computing a delay, so with Retries = N the multiplier never exceeds 2^(N-2). With the default Retries = 3 and a maxDelay above 2*baseDelay you only ever get baseDelay and 2*baseDelay, and the cap is unreachable. A smaller maxDelay is allowed, and then the cap bites earlier.

UseExponentialStrategy shortens each delay by a random amount of up to 10%. A mass failure (a downstream service is out) fails a whole batch at once, so without jitter every message would become ready again at the same instant and hand the recovering service the sharpest possible ramp. The jitter only ever shortens, so maxDelay stays a hard ceiling and the spread survives at the cap, where messages pile up the most.

AddInboxScheduler registers the inbox-scheduler health check (tag inbox-scheduler) itself, so the registration order does not matter. It reports Degraded when the last processing cycle is older than 2 × Period.

Note that ASP.NET Core maps Degraded to HTTP 200 by default, so a Kubernetes probe treats a stalled inbox as healthy. If you want a stalled worker to fail the probe, map it explicitly:

app.MapHealthChecks("/health/inbox", new HealthCheckOptions
{
    Predicate = registration => registration.Tags.Contains("inbox-scheduler"),
    ResultStatusCodes = { [HealthStatus.Degraded] = StatusCodes.Status503ServiceUnavailable },
});

Meter Inbox: ProcessCount, EmptyProcessCount, ProcessJobCount, ProcessJobSuccessCount, ProcessJobFailedCount, DeadLetteredCount, DuplicateCount, ExpiredBeforeStartCount, LeaseLostCount, ProcessDuration, plus two observable up/down counters: FreeJobCount (depth of what this service can handle) and DeadLetteredJobCount (buried messages this service is expected to review, never removed by cleanup).

ProcessJobFailedCount and DeadLetteredCount count what was actually written, not what was intended: when the lease is lost the outcome is not written at all, and counting the intent would page someone about a dead-lettered message that does not exist.

A steadily non-zero ExpiredBeforeStartCount or LeaseLostCount means the lease dies before the claimed batch is drained: raise lockTimeout, lower MessagesToProcess, or raise ConcurrencyLimit. The first counts leases that died while the message waited its turn, the second leases lost during processing itself. Neither spends an attempt: a lease running out is a sizing mistake, not a message failure, so the message returns to the queue untouched.

The body is serialized on enqueue and deserialized after a restart and after a deploy, exactly like the discriminator. System.Text.Json with default options is used, so the schema must stay backward compatible with whatever is still sitting in the table (up to CleanupOlderThan):

• Enums are written as numbers. Reordering or removing enum members silently changes the meaning of stored messages. Append new members at the end, or serialize enums as strings via a custom IInboxSerializer.
• Renaming a property, or changing its type, makes stored bodies unreadable: deserialization throws, the message is retried and then dead-lettered. Add new properties instead, and keep the old ones readable.
• Adding a property is safe: missing members deserialize to their default value.
• Property matching is case-sensitive by default.

Replace IInboxSerializer if you need different options; it is resolved from DI, so registering your own after AddInbox wins.

Content is stored as PostgreSQL text, and the library sets no upper bound on its length, unlike MessageId and ConsumerId, which are capped because they sit in the unique index. The inbox stores what it is handed. If the source is untrusted, bound the body size where the message enters the process: on the broker (RabbitMQ MaxMessageSize, Kafka message.max.bytes) or on the HTTP host (Kestrel MaxRequestBodySize). Deduplication does not help against this, because the MessageId is the sender's to choose, so a hostile sender can keep producing large bodies under fresh keys that live until CleanupOlderThan.

• PostgreSQL only (Dex.Cap.Inbox.Ef uses FOR UPDATE SKIP LOCKED and ON CONFLICT).
• Message types are discovered by reflection over loaded assemblies, so the assembly declaring IInboxMessage implementations must be loaded. In practice registering the handler loads it. The registry is built during host start (AddInbox registers a warm-up hosted service), so a duplicate or missing discriminator fails the host start rather than surfacing later inside the background worker. The discriminator itself is unrestricted: it reaches the claim SQL as a parameter, so a MassTransit urn:message:... or a nested type name is fine.
• That assembly must be loaded once. If the host puts it into several load contexts (a test runner, coverage instrumentation, a plugin host), the same message type exists twice as two distinct CLR types, and one discriminator maps to both. The inbox fails the start with AmbiguousMessageTypeException naming the load contexts instead of picking one: handlers are registered for a single type identity, so a guess would silently leave those messages unprocessed.
• Ordering between messages is not guaranteed; design handlers to be order-independent.

Guarantees that an operation is performed exactly once for a given idempotency key. On a repeat call the modification is skipped and the previously produced value (if any) is returned.

using Dex.Cap.OnceExecutor.Ef.Extensions;

services.AddOnceExecutor<AppDbContext>();
services.AddDefaultOnceExecutorScheduler<AppDbContext>(   // background cleanup of LastTransaction rows
    periodSeconds: 30,
    cleanupDays: 30);

// OnModelCreating
modelBuilder.OnceExecutorModelCreating();

// Remark: when using optimistic concurrency in PostgreSQL — exclude LastTransaction:
modelBuilder.UseXminAsConcurrencyToken(ignoreTypes: typeof(LastTransaction));

var executor = sp.GetRequiredService<IOnceExecutor<IEfTransactionOptions, AppDbContext>>();

var user = await executor.ExecuteAsync(
    idempotentKey: "create-user-42",
    modificator:  (db, ct) => db.Users.AddAsync(new User { Name = "Bob" }, ct).AsTask(),
    selector:     (db, ct) => db.Users.FirstOrDefaultAsync(x => x.Name == "Bob", ct));

Pass IEfTransactionOptions to control isolation level and command timeout:

await executor.ExecuteAsync(
    "key",
    (db, ct) => db.Users.AddAsync(new User { Name = "Bob" }, ct).AsTask(),
    options: new EfTransactionOptions { IsolationLevel = IsolationLevel.RepeatableRead });

Encapsulate idempotency, modification and read in a single class implementing IOnceExecutionStrategy<TArg, IEfTransactionOptions, TResult>:

public class CreateUserStrategy(AppDbContext db)
    : IOnceExecutionStrategy<CreateUserRequest, IEfTransactionOptions, string>
{
    public IEfTransactionOptions? Options { get; set; }
        = new EfTransactionOptions { IsolationLevel = IsolationLevel.RepeatableRead };

    public Task<bool> IsAlreadyExecutedAsync(CreateUserRequest arg, CancellationToken ct)
        => db.Users.AnyAsync(x => x.Name == arg.Name, ct);

    public async Task ExecuteAsync(CreateUserRequest arg, CancellationToken ct)
        => await db.Users.AddAsync(new User { Name = arg.Name, Years = 18 }, ct);

    public async Task<string?> ReadAsync(CreateUserRequest arg, CancellationToken ct)
        => (await db.Users.SingleOrDefaultAsync(x => x.Name == arg.Name, ct))?.Name;
}

services.AddStrategyOnceExecutor<CreateUserRequest, string, CreateUserStrategy, AppDbContext>();

var executor = sp.GetRequiredService<IStrategyOnceExecutor<CreateUserRequest, string>>();
var name = await executor.ExecuteAsync(new CreateUserRequest { Name = "Bob" }, ct);

Dex.Cap.Common.Ef ships two DbContext extension methods that wrap the operation in EF Core's IExecutionStrategy (transient-error retries) plus a transaction:

using Dex.Cap.Common.Ef.Extensions;

await db.ExecuteInTransactionAsync(
    operation: async ct =>
    {
        await db.Users.AddAsync(new User { Name = "Bob" }, ct);
        await outbox.EnqueueAsync(new OrderCreatedOutboxMessage { Args = "Bob" }, cancellationToken: ct);
        // SaveChangesAsync is called automatically before commit, but you may call it inside the operation
    },
    verifySucceeded: ct => db.Users.AnyAsync(u => u.Name == "Bob", ct),
    options: new EfTransactionOptions
    {
        IsolationLevel    = IsolationLevel.ReadCommitted, // default
        TimeoutInSeconds  = 60,                           // default
        ClearChangeTrackerOnRetry = true,                 // default
    },
    cancellationToken: ct);

Behaviour:

• On a transient NpgsqlException or TimeoutException the whole block is retried by IExecutionStrategy.
• verifySucceeded is called by the strategy to decide whether a retry is actually needed after an ambiguous failure.
• Throws UnsavedChangesDetectedException if the ChangeTracker already contains changes when the (root) call starts.
• Nested calls on the same DbContext instance participate in the existing transaction; a stricter isolation level than the outer one is rejected with InvalidOperationException.
• Nested calls on a different DbContext instance throw InvalidOperationException (silent atomicity loss is not allowed).

The discriminator (logical id of a message type, used both for serialization and for handler dispatching across services that share an outbox table) went through several iterations. Each iteration is a breaking change for consumers that used the previous shape.

1. Add static abstract to OutboxTypeId in every IOutboxMessage implementation:

   // before
   public string OutboxTypeId => "…";
   // after
   public static string OutboxTypeId => "…";

2. If you used AllowAutoPublishing, mark it static:

   public static bool AllowAutoPublishing => false;

3. Delete any custom IOutboxTypeDiscriminator / BaseOutboxTypeDiscriminator implementations and their DI registrations — they no longer exist.
4. If you wrote a custom auto-publishing handler (analogous to PublisherOutboxHandler<>), override static IsAutoPublisher => true. Regular per-type handlers should keep the default (false).
5. Bump straight to 8.2.2 or newer — 8.2.0 / 8.2.1 had short-lived constraints that were rolled back.