The Envelope That Did Not Cross

Every message broker worth using gives you at-least-once delivery, which is a polite way of saying it will deliver your message twice and let you sort it out. The standard answer is Idempotent Receiver: put a unique id on the message, record the ids you have processed, drop repeats. It is not subtle and it is not optional, and its entire cost is one identifier that has to survive the trip.

Part 6 traced the address. This part traces the envelope — everything that is about a message rather than in it — across Inflow's new process boundary. The short version: one field crosses, and it is not read.

What actually crosses

Inflow's publisher is one expression:

public Task SendAsync(IMessage message, Guid messageId, CancellationToken cancellationToken = default)
    => _publisher.PublishAsync(message, messageId.ToString("N"));

Convey's RabbitMqClient turns that into AMQP basic properties:

properties.MessageId = string.IsNullOrWhiteSpace(messageId)
    ? Guid.NewGuid().ToString("N")
    : messageId;
properties.CorrelationId = string.IsNullOrWhiteSpace(correlationId)
    ? Guid.NewGuid().ToString("N")
    : correlationId;
properties.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds());
properties.Headers = new Dictionary<string, object>();

So the wire carries: the payload, the message id Inflow supplied, a timestamp, an empty headers dictionary, and a correlation id. No content type naming the contract, no schema version, no trace id, no user id, no causation id.

Compare that with what the publishing code has in its hand two lines earlier, in MessageBroker.PublishAsync:

var requestId = _context.RequestId;
var traceId = _context.TraceId;
var userId = _context.Identity?.Id;
var correlationId = messageContext.Context.CorrelationId;

_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);

Five identifiers, gathered, logged, and then discarded one line later when SendAsync takes only two of them. Convey's PublishAsync has parameters for correlationId, spanContext, an arbitrary messageContext object and a headers dictionary. All four are left at their defaults.

The correlation id that is worse than missing

Look again at the CorrelationId line, because it does something that a missing field would not.

When no correlation id is supplied, Convey mints a fresh random Guid for every publish. So a correlation id does cross the boundary. It is well-formed, it appears in the RabbitMQ management UI, it appears in Convey's own “Handling a message ... Correlation ID: ...” log lines on the consuming side, and it is different for every message and related to nothing.

That is a genuinely nasty failure mode, and it is worth separating from the ordinary complaint about missing observability. An absent field is a gap you notice the first time you go looking. A present, plausible, always-unique field is a gap that answers your question with a wrong answer — you correlate two log lines, get nothing, and conclude the events were unrelated. Causal tracing does not degrade at this boundary; it inverts.

To be fair, this is Convey's default rather than Inflow's decision, and defaulting to something is defensible for a library that cannot know whether an ambient correlation id exists. But Inflow does have one, in messageContext.Context.CorrelationId, and IMessageBrokerClient.SendAsync has no parameter to carry it.

The underscore that is not the problem

The obvious suspect on the receiving side is this, from RabbitMqMessageSubscriber:

_busSubscriber.Subscribe<T>(async (serviceProvider, @event, _) =>
{
    using var scope = serviceProvider.CreateScope();
    await scope.ServiceProvider.GetRequiredService<IEventHandler<T>>().HandleAsync(@event);
});

A discarded third parameter, right where you would expect the message metadata to arrive. It reads like the whole story, and I want to be careful here, because it is not.

Convey's subscriber background service builds that third argument as _contextProvider.Get(args.BasicProperties.Headers), and ContextProvider.Get looks for a single header — message_context by default — deserialises it if present, and returns null otherwise. That header is only written when rabbitMq.context.enabled is true, and neither host on this branch has a context section at all. So the discarded _ is null on every delivery. Discarding it costs nothing.

The loss is somewhere less obvious, which is exactly why it is worth reading the dependency rather than assuming.

Where the id actually is

Immediately before invoking the handler, Convey does this:

var messagePropertiesAccessor = scope.ServiceProvider.GetRequiredService<IMessagePropertiesAccessor>();
messagePropertiesAccessor.MessageProperties = new MessageProperties
{
    MessageId = args.BasicProperties.MessageId,
    CorrelationId = args.BasicProperties.CorrelationId,
    Timestamp = args.BasicProperties.Timestamp.UnixTime,
    Headers = args.BasicProperties.Headers
};

MessagePropertiesAccessor is registered as a singleton and is backed by a static AsyncLocal. So by the time control reaches Inflow's lambda — and inside the scope it creates, and inside the handler that scope resolves — the message id that was published is sitting in ambient async-local state, one GetRequiredService<IMessagePropertiesAccessor>() away.

The identifier needed for Idempotent Receiver is put on the wire deliberately, delivered faithfully, parked in an accessor the handler can reach, and never asked for. That is the finding, and it is a nicer one than “they threw away the parameter”, because it means the distance between this branch and a deduplicating integration is smaller than it looks.

The cache that has to miss

Why doesn't the existing inbox machinery pick it up? Because it was built for a world where the message never left the heap.

public class MessageContextProvider : IMessageContextProvider
{
    private readonly IMemoryCache _cache;
    public IMessageContext Get(IMessage message) => _cache.Get<IMessageContext>(message);
}

The message id is stored in an IMemoryCache keyed on the message object itself, with a one-minute sliding expiry, written by MessageContextRegistry.Set(message, context) on the publishing side. In-process that is elegant: the same instance travels from publisher to handler, so the object is the key, and no envelope is needed.

A message deserialised off RabbitMQ is a brand-new instance that was never registered. The lookup cannot hit. And InboxEventHandlerDecorator — which does still apply on the RabbitMQ path, because the subscriber resolves IEventHandler<T> from the container — does this with the result:

var context = _messageContextProvider.Get(@event);
var name = @event.GetType().Name.Underscore();
await inbox.HandleAsync(context.MessageId, name, () => _handler.HandleAsync(@event, cancellationToken));

Get returns null on a miss, so context.MessageId on a wire message dereferences null. Nobody has hit that, because outbox.enabled is false in the Bootstrapper and the extracted service has no outbox section at all — so the decorator is never registered in either host. The whole path is latent.

Skip the dedup rather than read the id

Which makes the change the transition commit did make to EfInbox the most revealing four lines on the branch:

-        _logger.LogTrace($"Received a message with ID: '{messageId}' to be processed ('{module}').");
-        if (await _set.AnyAsync(m => m.Id == messageId && m.ProcessedAt != null))
-        {
-            _logger.LogTrace($"Message with ID: '{messageId}' was already processed ('{module}').");
-            return;
-        }
+        var saveToInbox = messageId != Guid.Empty;
+        if (saveToInbox)
+        {
+            // ... the duplicate check, now conditional
+        }

Read what that guard assumes. It expects messages to arrive with no usable id, and its response is to skip deduplication for them — to make the inbox a no-op rather than to go and find the identifier. The same commit added TransactionsDisabled to OutboxOptions so the surrounding transaction can be skipped too, and the two changes are the same admission: in-process, the dispatch boundary, the DI scope and the transaction boundary all coincided; after extraction they do not.

Against an at-least-once transport with three configured retries and — as Part 5's neighbour, the missing dead-letter configuration, shows — nowhere for a poison message to go, duplicate handling is unprotected in the shipped configuration, and the guard that would have caught it is switched off by the absence of the identifier it needed.

And it is worse in the extracted service, for a reason unrelated to any of this. MessageBroker only builds a MessageContext when message.GetModuleName() is non-empty, and that helper reads the namespace: Inflow.Services.Customers.Core.* is not a *.Modules.* namespace, so it returns empty string. The service therefore takes the Guid.NewGuid() from the top of the loop and publishes it, having registered nothing anywhere. Its message ids are unique, wire-visible, and correlated with no context on either side.

What it would have taken

I want to end fairly, because the gap here is genuinely small and the author clearly knew the direction of travel — the outbox, the inbox, the decorator, the registry and the id on the wire are all present.

• A correlationId parameter on IMessageBrokerClient.SendAsync, passed through to Convey. One parameter, two call sites.
• On receive, read IMessagePropertiesAccessor.MessageProperties.MessageId inside the subscriber lambda and call IMessageContextRegistry.Set(@event, new MessageContext(id, context)) before invoking the handler. Perhaps ten lines, and the existing inbox starts working unchanged.
• A null check in InboxEventHandlerDecorator so a missing context is a logged warning rather than a dereference.

That is the whole distance between this branch and a correlated, deduplicating integration: one parameter and about a dozen lines. The abstraction was made narrow enough to be transport-agnostic, and then nobody widened it back to carry what a real transport needs.

Next, extraction by copy and disable: two complete copies of the same domain in one repository, and the boolean that decides which one runs.