The Bus That Never Leaves the Process

Part 7 followed a message through the translator. This part is about how it gets there — the one class every handler in the estate publishes through, and what it does with what you give it.

InMemoryMessageBroker is eighty-eight lines and implements a two-method interface. A module handler that has just changed some state writes one line:

await _messageBroker.PublishAsync(new CustomerCompleted(customer.Id, customer.Name,
    customer.FullName, customer.Nationality));

No topic, no exchange, no routing key, no channel. IMessageBroker is a genuinely textbook Messaging Gateway: two PublishAsync overloads, one for a single message and one for an array, and zero infrastructure vocabulary leaking into module code. Every module in the estate depends on this interface and nothing below it.

The three-way branch

The private method both overloads funnel into is where the design lives:

private async Task PublishAsync(CancellationToken cancellationToken, params IMessage[] messages)
{
    // ... null filtering, then per-message context minting and logging ...

    if (_outboxBroker.Enabled)
    {
        await _outboxBroker.SendAsync(messages);
        return;
    }

    var tasks = _messagingOptions.UseAsyncDispatcher
        ? messages.Select(message => _asyncMessageDispatcher.PublishAsync(message, cancellationToken))
        : messages.Select(message => _moduleClient.PublishAsync(message, cancellationToken));

    await Task.WhenAll(tasks);
}

Three destinations, chosen by two configuration flags:

1. Outbox enabled — the messages go to IOutboxBroker, which writes rows to a database table and returns. Delivery becomes a background poller's problem. Shipped as false.
2. Async dispatcher on — the messages go into an in-memory Channel, and a BackgroundService drains it. Shipped as true.
3. Neither — ModuleClient.PublishAsync runs inline, on the caller's thread, inside the caller's await.

That third case matters more than it looks, and I will come back to it. First, the shipped path.

What the shipped path actually guarantees

Trace one event end to end with outbox:enabled: false and messaging:useAsyncDispatcher: true, which is the configuration in both appsettings.json and appsettings.test.json.

A command handler updates an aggregate and calls its repository's UpdateAsync, which calls SaveChangesAsync. The database transaction commits. Then the handler calls _messageBroker.PublishAsync(...). The broker mints a MessageContext, logs a structured line, and hands the message to AsyncMessageDispatcher, which writes an envelope to a Channel.CreateUnbounded<MessageEnvelope>() and returns. The handler returns. The HTTP response goes out — 200, or 204, or whatever the controller decided. Some time later, a background service reads the envelope and publishes it to the receivers.

Commit, then publish, with an in-memory queue in between and no durability anywhere. If the process dies between the commit and the drain, the state change survives and the event does not. This is at-most-once delivery, and it is the failure mode the outbox pattern exists to remove — which is why the outbox exists in this repository at all, and why part 11 is about it not working.

For a course demonstrating module decoupling, at-most-once is a completely reasonable choice, and it is announced honestly by outbox:enabled: false sitting in a config file anyone can read. What is worth being precise about is that the async dispatcher is not a smaller guarantee than the inline path — it is a different one. Inline publishing (branch three) means a receiver's exception propagates back into the publishing command handler, after its commit. The HTTP caller gets a 500 while the state change is durable and some receivers have already applied their side effects. Asynchronous publishing converts that into a logged error nobody is waiting on. Neither is right; they are differently wrong, and which one you want depends on whether your callers can tolerate a lie.

The context side-table

Before dispatching, the broker does something that shapes the rest of the framework:

var messageContext = new MessageContext(Guid.NewGuid(), _context);
_messageContextRegistry.Set(message, messageContext);

A fresh MessageId plus the ambient IContext — request id, correlation id, trace id, identity — are packaged and stored. Not on the message: IMessage is an empty marker interface and stays that way. The context goes into a side table:

public void Set(IMessage message, IMessageContext context)
    => _cache.Set(message, context, new MemoryCacheEntryOptions
    {
        SlidingExpiration = TimeSpan.FromMinutes(1)
    });

IMemoryCache, keyed on the message object itself. This is an inverted Envelope Wrapper, and as a technique it is defensible: domain records stay free of infrastructure fields, IMessage stays a marker, and any component that needs the context asks a provider for it. The logging decorators do exactly that.

Two consequences, both worth knowing before you copy the idea.

The key is a record, so value equality applies. Inflow's events are positional records — internal record CustomerVerified(Guid CustomerId) : IEvent; — and IMemoryCache uses EqualityComparer<object>.Default, which for a record means structural equality. Two CustomerVerified events for the same customer id, published within a minute of each other, are the same key. They share one cache entry and therefore one MessageId. Records changed what “the same object” means in C#, and every object-keyed cache in the ecosystem inherited that change; this is what it looks like when it lands.

The one-minute sliding expiration is a deadline. Anything that reads the context more than a minute after the last read gets null. On the shipped path that is fine — the drain happens in milliseconds. On the outbox path it is not: a row sitting unsent for two minutes has lost its context by the time the poller republishes it, and EfOutbox.SaveAsync dereferences context.MessageId without a null check.

The channel and its single reader

private readonly Channel<MessageEnvelope> _messages = Channel.CreateUnbounded<MessageEnvelope>();

Unbounded. There is no backpressure anywhere in this system: AsyncMessageDispatcher.PublishAsync awaits Writer.WriteAsync, which on an unbounded channel never blocks. If receivers are slower than publishers, the queue grows until the process runs out of memory. Backpressure you did not design is memory you did not budget. A BoundedChannelOptions with a FullMode — Wait for pushback, DropOldest for a lossy telemetry-style bus — is a two-line change that turns an unbounded failure into a chosen one.

The reader is one BackgroundService, and the whole of it is:

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    _logger.LogInformation("Running the async dispatcher.");
    await foreach (var envelope in _messageChannel.Reader.ReadAllAsync(stoppingToken))
    {
        try
        {
            _contextAccessor.Context ??= envelope.MessageContext.Context;
            await _moduleClient.PublishAsync(envelope.Message, stoppingToken);
        }
        catch (Exception exception)
        {
            _logger.LogError(exception, exception.Message);
        }
    }

    _logger.LogInformation("Finished running the async dispatcher.");
}

Four things about seventeen lines.

One reader, sequentially. Messages are drained in order, one at a time, and each publish fans out to its receivers with Task.WhenAll inside ModuleClient. That gives you global ordering across the whole estate — every message, every module — which is a much stronger guarantee than anyone needs and a much lower throughput ceiling than anyone wants. Ordering per aggregate is the useful property; ordering per process is what you get.

A handler failure is a logged line and nothing else. No retry, no backoff, no dead letter, no counter, no metric. The message is gone. Combined with the unbounded channel and the absence of any metrics surface in the repository — no ActivitySource, no System.Diagnostics.Metrics, no health endpoint — a failing receiver is invisible unless somebody is reading logs.

The try is inside the loop, not around the enumeration. When stoppingToken fires during shutdown, ReadAllAsync throws OperationCanceledException from the await foreach itself, outside the try. Everything still buffered in the unbounded channel is discarded. Nothing calls Writer.Complete() anywhere in the codebase, so the loop can only ever end by cancellation, which means the trailing "Finished running the async dispatcher." line is unreachable — a log message that exists to tell you about a graceful drain that cannot happen. Draining on shutdown is Writer.Complete() in StopAsync plus a try around the enumeration; roughly six lines.

And ??= on line 34. That one deserves its own article, because it converts every trace in the process into a copy of the first one, and I have left it until last on purpose.

The logging is the good half

I have spent most of this article on what the bus does not guarantee, so let me be precise about what it does exceptionally well: it tells you everything.

The broker logs before it dispatches, and the log line is fully structured:

_logger.LogInformation("Publishing a message: {Name} ({Module}) [Request ID: {RequestId}, Message ID: {MessageId}, Correlation ID: {CorrelationId}, Trace ID: '{TraceId}', User ID: '{UserId}]...",
    name, module, requestId, messageId, correlationId, traceId, userId);

Seven named properties, not an interpolated string — so Serilog emits them as fields, and Seq or any structured sink can filter on any of them. name comes from message.GetType().Name.Underscore(), giving customer_completed rather than a CLR type name, which is a small and correct choice: the log is a description of a domain fact, not of a class.

The same shape appears in all three logging decorators, before and after each handler. So a single event produces “publishing”, then per receiver “handling” and “handled”, each carrying the module label and the same message id. Reconstructing a fan-out from logs alone is genuinely possible here, which is more than I can say for most in-process buses, where the message vanishes into a mediator and reappears as whatever the handler decided to log.

Two blemishes, both cosmetic and both worth noticing because of what they say about review. The message template ends User ID: '{UserId}] — an opening quote with no closing quote, and a stray bracket where the closing one should be. LoggingEventHandlerDecorator has the identical malformation, and LoggingCommandHandlerDecorator's two lines disagree with each other on the same punctuation: one ends '{UserId}]'... and the other '{UserId}']. Four templates, three different endings, one of them consistent by accident. Structured logging means nobody reads the template as text, so a broken one survives indefinitely — which is exactly the trade you accept when you stop formatting strings.

Which makes the next finding worse rather than better, because a log this detailed is a log people will trust. Next, one assignment freezes every trace — the ??= on line 34 of the dispatcher, and why every one of those correlation ids is the same one after the first message.