MessagingKit.Outbox 0.4.0

MessagingKit

Transactional outbox and inbox for .NET 10, on EF Core and PostgreSQL.

Reference it from a module and that module can send and receive durably. Messages between modules in one host travel over the same outbox → transport → inbox seam they would use across a network, so moving a module into its own deployment later changes one registration and no module code.

The framework knows nothing about brokers, HTTP, or any other transport. It owns durability and the state machine; you implement IMessageTransport and plug in whatever moves the bytes. An in-process transport ships in the box.

The problem

You cannot atomically write to your database and talk to another system:

db.Invoices.Add(invoice);
await db.SaveChangesAsync();     // committed
await broker.PublishAsync(msg);  // process dies here — invoice exists, nobody was told

Swap the order and you get the opposite bug: the message goes out, the transaction rolls back, and you've announced an invoice that does not exist.

The outbox closes that gap by making the intent to send part of the same transaction. The inbox closes the matching gap on the receiving side, where a retried delivery would otherwise be processed twice.

Outbox means don't lose it. Inbox means don't do it twice. A broker replaces neither — it only carries the message in between.

Packages

Take MessagingKit for the whole thing, or the halves on their own if you only send or only receive. Each package is one namespace, so one using per package brings in everything it exposes: the contracts (IMessageHandler<T>, [Message], AddMessageHandler, MessagingKitInstrumentation) are in MessagingKit.Abstractions, and the outbox's IOutbox, OutboxMessage and OutboxOptions are all in MessagingKit.Outbox.

Modules

A module registers what it handles. Nothing else in the host needs to know:

A module never names the host's DbContext — it only declares what it sends and handles, so it depends on MessagingKit.Abstractions alone:

// inside the email module
public static IServiceCollection AddEmailModule(this IServiceCollection services) =>
    services.AddMessageHandler<SendEmail, SendEmailHandler>();

// inside the billing module — sends, does not handle
services.AddMessageContract<SendEmail>();

The host owns the context, declares it once, and wires the modules:

builder.Services.AddMessaging<AppDbContext>(builder.Configuration);

builder.Services.AddEmailModule();
builder.Services.AddBillingModule();

Order does not matter — both sides only add to the service collection.

For a single-module host the fluent form does the same thing in one statement:

builder.Services.AddMessaging<AppDbContext>(builder.Configuration)
    .Handles<SendEmail, SendEmailHandler>();

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.AddMessaging();   // messaging.outbox + messaging.inbox
}

The tables land in your DbContext, so dotnet ef migrations add picks them up in your own migration history — the package ships no migrations of its own.

AddMessaging is safe to call once per module. The dispatcher and processor are registered once no matter how many modules call it, and in-process delivery is the default transport.

Billing then sends inside its own transaction, and the email module handles it:

db.Invoices.Add(invoice);
outbox.Add(new SendEmail(customer.Email, "Your invoice"));
await db.SaveChangesAsync();    // both rows commit, or neither does

Message names

A message carries a name on the wire. It is derived from the type — SendEmail becomes send-email, HTTPRequest becomes http-request — so Handles<TMessage, THandler>() registers the same name for the sender and the handler and they cannot drift apart.

Renaming the class renames the message, which orphans rows already queued. Pin the name on anything in production:

[Message("email.send.v2")]
public record SendEmail(string To, string Subject);

The string overloads are still there when you want to be explicit: AddMessage<T>("name"), AddHandler<T, THandler>("name").

Transports and routing

In-process delivery is the default. Route specific types elsewhere as modules move out of the host:

services.AddMessaging<AppDbContext>(config)
    .Handles<SendEmail, SendEmailHandler>()          // stays local
    .UseTransportFor<BrokerTransport, InvoiceIssued>();  // this one goes to the broker

Or set a different default and keep only some types local:

.UseInProcessTransport("send-email")   // named types stay in the host
.UseTransport<BrokerTransport>()       // everything else goes to the broker

A routing key matches either a message's destination or its type, so a module can address another module directly and override the type's usual route:

outbox.Add(new SendEmail(...), destination: "email-module");

.UseTransport<BrokerTransport>("email-module")   // everything addressed there
.UseInProcessTransport("send-email")             // everything of this type

Resolution runs destination first, then message type, then the default, then any IMessageTransport registered directly in DI. A message with no route at all is dead-lettered immediately rather than retried — a missing registration is a wiring bug, and retrying it for ten attempts only buries the cause.

Implementing a transport is one method:

public sealed class BrokerTransport(IBrokerClient broker) : IMessageTransport
{
    public async Task<Result> SendAsync(MessageEnvelope envelope, CancellationToken ct = default)
    {
        try
        {
            await broker.PublishAsync(envelope.Destination ?? envelope.Type, envelope.Payload, ct);
            return Result.Success();
        }
        catch (BrokerException ex)
        {
            return Error.Transient("broker.unreachable", ex.Message);
        }
    }
}

Transports return a ResultingKit Result. Only an ErrorKind.Transient error is retried; return any other kind (Error.Validation, Error.Failure, …) for failures retrying cannot fix — a malformed destination, a rejected payload. Those skip the retry ladder and go straight to Dead.

Receiving

Handlers take the deserialized message plus a MessageContext carrying the message id, type, attempt count, and headers:

public sealed class SendEmailHandler(IEmailClient client) : IMessageHandler<SendEmail>
{
    public Task HandleAsync(SendEmail message, MessageContext context, CancellationToken ct = default) =>
        client.SendAsync(message.To, message.Subject, ct);
}

With InProcessTransport the outbox hands messages to the inbox for you. When something else receives them — a broker subscriber, an HTTP endpoint — call the inbox yourself:

if (await inbox.TryStoreAsync(envelope, ct))
{
    // stored; the processor will handle it
}
// false means already seen — ack and move on

TryStoreAsync is the deduplication point: the message id is the primary key, so a redelivery cannot create a second row. Acknowledge the transport immediately; the background InboxProcessor resolves IMessageHandler<T>, runs it, and marks the row Processed. A throwing handler is retried with backoff, then dead-lettered.

The inbox runs a handler once per message, but a retry runs it again after a side effect may already have happened — an HTTP call that succeeded but whose response was lost. When the handler calls something that accepts an idempotency key, derive the key from context.MessageId: it is the same on every attempt and unique per message. MailingKit does exactly this for Resend.

À la carte

Outbox and inbox work on their own if you do not want both:

builder.Services.AddOutbox<AppDbContext>(builder.Configuration)
    .AddMessage<SendEmail>()
    .UseTransport<BrokerTransport>();

builder.Services.AddInbox<AppDbContext>(builder.Configuration)
    .AddHandler<SendEmail, SendEmailHandler>();

modelBuilder.AddOutbox();   // messaging.outbox
modelBuilder.AddInbox();    // messaging.inbox

Telemetry

MessagingKit instruments itself the way ASP.NET Core and HttpClient do: spans on an ActivitySource, metrics on a Meter from the container's IMeterFactory, and nothing but System.Diagnostics underneath. No OpenTelemetry package is referenced; your host opts in by name:

using MessagingKit.Abstractions;

builder.Services.AddOpenTelemetry()
    .WithTracing(t => t.AddSource(MessagingKitInstrumentation.ActivitySourceName))
    .WithMetrics(m => m.AddMeter(MessagingKitInstrumentation.MeterName));

Names follow the current OpenTelemetry messaging semantic conventions where the meaning matches; everything MessagingKit-specific is under messagingkit.. messaging.system is always messagingkit.

Tracing

Trace context is captured when a message is staged and carried in its headers, so the send and the handling that follows belong to the caller's trace rather than two unrelated ones.

Both are parented to the context captured at staging. The send span is a Client rather than a Producer because it is not that creation context, which the conventions reserve Producer for. Spans are named by the message type: the conventions name them by destination, but a MessagingKit destination is whatever the caller passed, and span names must stay low-cardinality.

A failed attempt also sets the span status to Error with the failure message, and a thrown exception is recorded as an exception event.

Metrics

The first five carry messagingkit.message.type (the wire name, as on the spans) and messagingkit.outcome:

The four semantic-convention instruments also carry messaging.operation.name and messaging.operation.type (send for the outbox, process for the inbox), and failed attempts carry error.type:

• a transport's Error.Code when it returns a failed Result — keep codes stable and low-cardinality, as ResultingKit asks;
• the exception's full type name when one is thrown (the handler's own exception, not the reflection wrapper);
• unknown_message_type or no_handler when the inbox cannot dispatch a row at all.

Message ids and destinations are never metric tags — ids are unbounded, and a destination is whatever a caller passed. Durations are measured with Stopwatch, not the injected TimeProvider, so they stay real when tests fake the clock. Histograms advise the OpenTelemetry messaging buckets (5 ms to 10 s).

Backlog. The two messages instruments report the last count the dispatcher or processor took on its own loop, at most once every BacklogSampleIntervalSeconds (default 30, 0 turns it off), and only while something is subscribed to them. Collection reads the cached count and never touches the database; nothing is reported until the first count. A failed count is logged as a warning and the previous one kept. Every replica counts the same shared table, so aggregate these across instances with max, not sum.

Compile-time checks

MessagingKit.Abstractions ships an analyzer, so referencing it is all it takes:

MK1002 is a warning rather than an error because the registration may legitimately live in another project; suppress it there. Turn on MK1004 once messages are in production and a class rename would orphan queued rows:

dotnet_diagnostic.MK1004.severity = warning

Startup validation

A message delivered in-process with no handler registered fails the host at boot:

MessagingKit is misconfigured:
  - 'send-email' is delivered in-process but no IMessageHandler<SendEmail> is registered.
    Register a handler with Handles<SendEmail, THandler>(), or route it to another transport.

That is a failed deploy instead of a dead-lettered row discovered days later. Messages routed to another transport are skipped — they are handled wherever they land.

Configuration

Both halves bind the same shape, under Outbox and Inbox:

Operational notes

• Delivery is prompt, not instant. Committing a transaction that staged outbox rows wakes the dispatcher, and storing to the inbox wakes the processor, so a message does not wait out the poll interval. The interval is the fallback — a missed signal costs latency, never correctness.
• Ordering is not guaranteed. Messages dispatch in parallel; if you need per-key ordering, that is a transport concern.
• Delivery is at-least-once. A process can die after the transport accepts a message but before the row is marked Sent. That is precisely why the receiver needs an inbox.
• LockDurationSeconds must exceed your slowest send, or a second worker reclaims a row still in flight and the message goes out twice.
• Both tables grow forever. Add a retention job that deletes Sent and Processed rows past your audit window; keep Dead until someone has looked at them.
• One inbox table per host, shared by every module and routed by message type. A module gets its own inbox when it gets its own database.
• Multiple replicas are safe — SKIP LOCKED guarantees a row is claimed by exactly one worker.

Testing against it

MessagingKit.Testing drives both halves to completion so a test asserts on the effect of a message rather than sleeping until the background loops happen to run:

outbox.Add(new SendEmail("ada@example.com", "Your invoice"));
await db.SaveChangesAsync();

await services.DrainMessagingAsync();

Assert.AreEqual(1, mailbox.Sent.Count);

It drives the same dispatcher and processor production uses, so what the test exercises is what ships. DrainOutboxAsync() and DrainInboxAsync() run one half when you want to assert on what was delivered before anything handles it.

Tests

dotnet test                                                     # every kit
dotnet test kits/messaging-kit/tests/MessagingKit.UnitTests     # no Docker needed
dotnet test kits/messaging-kit/tests/MessagingKit.IntegrationTests   # needs Docker

MessagingKit.UnitTests covers naming, registration, signalling, backlog sampling, startup validation, and the analyzer — no infrastructure, so it runs in about a second and parallelises.

MessagingKit.IntegrationTests runs against a real PostgreSQL container using TestingKit, with a fake in-memory transport — the framework is transport-agnostic, so the tests need no broker. It is [assembly: DoNotParallelize]: the suite shares one database and resets between tests.

Test doubles shared by both live in MessagingKit.TestSupport so they cannot drift apart.