0063-failed-delivery-context

title

Failed Delivery Context for Confirmation-Based Producers

status

Accepted

author

Brighter Team

created

2026-06-10

summary

Enriches the ISupportPublishConfirmation.OnMessagePublished event to carry a PublishConfirmationResult (wire topic, original publish ActivityContext, message id) so OutboxProducerMediator can emit a linked confirmation span, log a warning, and trip the circuit breaker on async publish failures from Kafka and RMQ producers.

0063. Failed Delivery Context for Confirmation-Based Producers

Date: 2026-06-10

Status

Accepted

Context

Confirmation-based producers (those implementing ISupportPublishConfirmation, currently KafkaMessageProducer and RmqMessageProducer in both the RMQ.Async and RMQ.Sync projects) report publish success or failure asynchronously, after the synchronous Send/SendAsync call has already returned. They do this by raising the OnMessagePublished event (src/Paramore.Brighter/ISupportPublishConfirmation.cs:42), whose signature today is event Action<bool, string> (success, messageId).

OutboxProducerMediator wires a single handler onto that event, once per producer, in its constructor via ConfigureCallbacks (src/Paramore.Brighter/OutboxProducerMediator.cs:162, iterating _producerRegistry.Producers). The handlers (ConfigureAsyncPublisherCallbackMaybe at :737 and ConfigurePublisherCallbackMaybe at :764) handle only the success == true branch: they log "sent" and call MarkDispatched(Async) (:749, :776). The success == false branch is silently dropped. The result is the four gaps in the parent requirement: no log, no telemetry, no circuit-breaker trip, and (for Kafka) a discarded message id and a swallowed ProduceException.

Contrast the non-confirmation path: in DispatchAsync, a synchronous send failure is observed inline via the sent boolean and trips the breaker with if(!sent) TripTopic(message.Header.Topic); (:998); the batch path trips TripTopic(batch.RoutingKey) (:933). Confirmation-based producers never reach that code because their send returns "successfully" and the real result arrives later on the callback.

Three facts shape the design and create dependencies:

C-6 (verified). The callback closure captures a single construction-time RequestContext produced by requestContextFactory.Create() (:162). It is empty of per-publish data: it carries neither the per-message wire topic nor the per-publish producer span. The closure does not capture the per-message message. So neither the topic nor the original trace context for FR-1/2/3 is reachable from captured closure state. (The captured context is not wholly unused: the success branch passes it to MarkDispatched/MarkDispatchedAsync.)
C-7 (trace-context reachability). The original publish span is created in Dispatch/DispatchAsync via _tracer.CreateProducerSpan(...) (:825, :970) and ended in the finally via _tracer.EndSpans(producerSpans) (sync :863, per-message async :1006 / BrighterTracer.EndSpan at src/Paramore.Brighter/Observability/BrighterTracer.cs:709). By the time the asynchronous confirmation fires, that span has already ended (C-2). Its ActivityContext must be made reachable at callback time to build FR-2's ActivityLink.
C-8 (wire-topic source). FR-1/FR-3 must use the per-message wire topic message.Header.Topic, the exact value the non-confirmation path trips (:998), not producer.Publication.Topic. The two diverge for reply/rewritten-topic messages: GetProducerLookupTopic (:786-808) reads a ProducerTopicHeaderName bag entry that points lookups at the static topic while Header.Topic holds a dynamic reply address. The wire topic is not carried by the current Action<bool, string> delegate, so it too must be plumbed to the callback.

A further wrinkle constrains how the context is captured: spans are thread-affine. RequestContext.Span is thread-keyed by ManagedThreadId (src/Paramore.Brighter/RequestContext.cs:113-125 — backed by a ConcurrentDictionary<int, Activity> _spans keyed on Thread.CurrentThread.ManagedThreadId), and producer.Span is a single mutable property on a shared producer, nulled right after the send (:858) and reused per message. The confirmation callback runs on a different thread from Dispatch, later in time (Kafka marshals via Task.Run at KafkaMessageProducer.cs:373, 381; RMQ invokes from its broker nack handler at RmqMessageProducer.cs:480). Therefore none of these can be re-read at callback time to recover S1 — the value of the ActivityContext (a struct, valid after the span ends) must be captured synchronously at dispatch time and carried explicitly to the callback. The Decision sources that value from Activity.Current (set to S1 by CreateProducerSpan and AsyncLocal, so race-free across concurrent dispatch), not the shared producer.Span field (which would race) and not RequestContext.Span (which holds the parent span).

Parent Requirement: specs/0034-failed-delivery-context/requirements.md

Scope. This single ADR covers, together: the callback-plumbing/contract mechanism (how the wire topic and original trace context reach the callback); the standalone confirmation-span + ActivityLink observability design, emitted on both the success and failure branches (FR-2) — which, on success, also re-parents the previously orphaned MarkDispatched Outbox span (C-6) so it correlates to the publish; the breaker-trip wiring on the false branch (FR-3); the Kafka publisher logging of the swallowed ProduceException (FR-6) and id propagation through PublishResults (FR-8); RMQ verification (FR-9) and RMQ's raise-site context enrichment (FR-2); error isolation (NFR-4 / AC-14); and — as a PO-authorised scope expansion (2026-06-11) — making InMemoryMessageProducer a confirmation producer with an opt-in async-confirm + failure-injection capability so the whole flow is exercisable in-process with a production provider rather than a test-only fake. This is a direct application of ADR 0012 — Provide In-Memory and Simple Implementations: Brighter deliberately ships full-featured in-memory providers (not fakes), with their own tests, that end users may run — including as real providers where simplicity is valued over reliability (e.g. an in-memory bus in a modular monolith, or early in development before moving to persistent storage). Bringing InMemoryMessageProducer to confirmation parity with Kafka/RMQ keeps that provider full-featured rather than letting it lag the abstraction. The success-branch confirmation span is a deliberate (PO-approved 2026-06-11) symmetric expansion, justified as a cleanup of how the confirmation callback reports rather than a new success-path feature — the success branch's delivery semantics ("sent" log + mark-dispatched) are unchanged, and it gains no Warning and no breaker trip. Out of scope: Sweeper/retry semantics (C-1, OOS-1), the dead wrong-typed catch (ProduceException<string, string>) clauses at KafkaMessageProducer.cs:262 and :336 (OOS-2), and the success == true delivery behaviour (OOS-4, now narrowed to delivery semantics only — the success-branch confirmation span is in scope). Note on OOS-3/NFR-5: those forbid a toggle for the failed-delivery observability/breaker behaviour (which stays always-on); the new InMemory switch toggles only the in-memory provider's confirm timing/outcome for local dev + tests, not the mediator behaviour, so it does not contradict OOS-3/NFR-5.

Decision

Adopt mechanism (a): extend the contract — but only the one contract that must change, the confirmation contract (ISupportPublishConfirmation.OnMessagePublished). The original publish span's ActivityContext is sourced by the producer from Activity.Current (which CreateProducerSpan sets to the producer span S1 immediately before Send is called, and which is AsyncLocal so it is race-free across concurrent dispatch), so no change to the IAmAMessageProducerSync/Async Send signatures is required.

Producer self-sources the publish span (no Send signature change). When OutboxProducerMediator.Dispatch/DispatchAsync create the producer span S1 via CreateProducerSpan, that helper also sets Activity.Current = S1 (BrighterTracer.cs:700). The producer therefore captures the value Activity.Current?.Context (an ActivityContext — a struct that stays valid after the span ends) synchronously, as the first action inside Send/SendAsync, before any await, and carries it alongside the per-message state through its existing confirmation machinery (see Implementation Approach) to the point where it raises OnMessagePublished. Activity.Current is AsyncLocal, so concurrent dispatches on different threads see independent values — sourcing the span this way is race-free under concurrent dispatch to the same producer, unlike reading the shared producer.Span property (see "Capturing the context at dispatch" for why the shared field is rejected). No change to the producer Send signatures is required (the rejected Send-parameter approach is in Alternatives Considered).
Enrich the OnMessagePublished contract so the callback receives, in addition to success + message id, the per-message wire topic (message.Header.Topic) and the original publish ActivityContext (the value captured in step 1). The producer holds both pieces — it already receives the message in Send/SendAsync and self-sources the span from Activity.Current — so the same enrichment carries both. This is the only interface contract that changes, and it changes only for the producers that implement ISupportPublishConfirmation: Kafka, RMQ, and — newly, as an opt-in capability (see "Confirmation-capable in-memory producer" below) — InMemoryMessageProducer.

OutboxProducerMediator's callback then emits a standalone short-lived confirmation span (S2) on every invocation — success and failure — via BrighterTracer, carrying an ActivityLink to the captured ActivityContext when present, or no link when absent (FR-2, degraded per AC-2b), started and stopped synchronously within the callback (NFR-2). S2 becomes the ambient parent for whatever the callback does next, which differs by branch:

On success == true:

keeps the existing Info "sent" log and the MarkDispatched/MarkDispatchedAsync Outbox call (delivery semantics frozen, OOS-4);
the MarkDispatched DB span is explicitly re-parented to S2 (by giving MarkDispatched(Async) a RequestContext whose Span == S2), so it nests under S2 (which links to S1), fixing today's orphaned/null-parent Outbox span (C-6) — see "Re-parent the MarkDispatched DB span EXPLICITLY" below; no Warning, no breaker trip (AC-13).

On success == false:

logs at Warning including message id (empty-marked when blank) and wire topic (FR-1, NFR-1);
trips the breaker via the existing private TripTopic(RoutingKey?) (:1168) with the wire topic (FR-3, NFR-6), giving exact parity with :998;
does not mark dispatched, does not await any Outbox/broker operation, and does not bubble (FR-4, AC-12b).

On both branches the observability work is wrapped so a throw there cannot escape the callback (NFR-4 / AC-14).

Choosing the enriched `OnMessagePublished` shape

The existing event is event Action<bool, string> OnMessagePublished. Changing its delegate type is both source- and binary-breaking for implementors — every producer raising it, plus the mediator's two handlers, must be recompiled and updated, and a precompiled third-party assembly referencing the old delegate type will fail to bind — but it is invisible to end users (gateway authors raise it; application code never does). Because only ISupportPublishConfirmation implementors (Kafka, RMQ, and now InMemoryMessageProducer) are affected, the blast radius is the confirmation producers, not all gateways. (InMemory's current OnMessagePublished is an orphaned Action<bool, Id> with no external subscribers — only raised internally at InMemoryMessageProducer.cs:100, 123, 145 — so adopting the enriched delegate has no in-repo subscriber-migration cost and tidies a latent inconsistency.) Two candidate shapes:

A wider tuple delegate, e.g. Action<bool, string, RoutingKey?, ActivityContext?>. Rejected: positional bool/string/nullable args invite mis-wiring, are not self-describing, and adding a fifth datum later breaks the signature again. This also conflicts with the design principle against primitive obsession.
A single immutable args record carried by Action<PublishConfirmationResult> (chosen). Under Responsibility-Driven Design the result type is an information holder with named, intention-revealing members:

public sealed record PublishConfirmationResult(
    bool Success,
    Id MessageId,                 // Brighter Id; Id.Empty for the unknown/missing case (FR-5)
    RoutingKey? Topic,            // message.Header.Topic — the wire topic (C-8)
    ActivityContext? PublishSpanContext);  // original publish span (C-7); null => Link degrades (FR-2)

MessageId is typed Id (Brighter's id type, Id.cs:38) rather than string: Id.Empty (:53) gives FR-5 its explicit unknown/missing marker, the implicit string ↔ Id conversions (:95, :102) let Kafka/RMQ keep passing their raw string ids unchanged, and InMemoryMessageProducer already holds the id as Id. This keeps the contract stable against future additions (add a property, not a parameter), makes the degradation contract explicit (PublishSpanContext == null ⇒ no link), and reads at the call site. PublishConfirmationResult is a new public type in the Paramore.Brighter namespace (alongside ISupportPublishConfirmation); it is part of the public API surface, so it carries the project's XML-documentation obligation on the type and every member, and it participates in semantic versioning as an additive public type.

Both branches use the record, and both consume all four members. OnMessagePublished is raised on success and failure, so its single argument carries both cases. The producer populates all four members on both raises: Success, MessageId, Topic (message.Header.Topic, which it holds), and PublishSpanContext (the value captured in step 1, or default when no span was active). The mediator now reads PublishSpanContext on both branches to build the confirmation span's ActivityLink (FR-2 is symmetric); the success branch additionally still reads MessageId to call MarkDispatched. Topic is used for the failure-branch log/breaker and as a span tag on both branches. Requiring all four on both raises makes the contract unambiguous so the three implementors populate it identically.

Backward-compat story, stated honestly: this is a source- **and binary-**breaking change to a public interface, but only for implementors of ISupportPublishConfirmation (Kafka, RMQ, and any third-party confirmation producer); they must update their raise sites and recompile. End users are unaffected, and non-confirmation gateways are untouched. We accept this because the data simply cannot reach the callback without it (C-6/C-7/C-8), and a correlation map or header round-trip (see Alternatives) carry worse trade-offs.

Architecture Overview

Dispatch / DispatchAsync (mediator)                     OnMessagePublished callback (later, other thread)
  S1 = tracer.CreateProducerSpan(...)  ── publish span ──┐
       └─ sets Activity.Current = S1   (BrighterTracer :700)
  producer.SendAsync(message) ───────────────────────────┤
    └─ inside Send (sync, first line, pre-await):         │
         ctx = Activity.Current?.Context  (value, == S1)  │  ← AsyncLocal, race-free
         stash ctx with per-message confirmation state    │
  finally: tracer.EndSpans(...)   << S1 ENDED >>          │
                                                          │   broker ack/nack arrives (async)
                                                          ▼
                                      producer raises OnMessagePublished(
                                          PublishConfirmationResult(
                                              Success = true | false,
                                              MessageId=id,
                                              Topic=message.Header.Topic,        (C-8)
                                              PublishSpanContext=ctx ))          (C-7, may be null)
                                                          │
                                                          ▼
                                      mediator handler — emit S2 on EVERY confirmation:
                                        S2 = standalone confirmation span,            (FR-2)
                                          links=[ new ActivityLink(ctx) ] if ctx!=null
                                          (S1 NOT reopened; unrelated Activity.Current NOT used)
                                          start + stop within callback                (NFR-2)
                                          S2 is the ambient parent for the work below
                                          │
                            success ──────┤────── failure
                              │                      │
        Log.Info("sent")      │                      │   Log.Warning(id-or-"unknown", topic)   (FR-1)
        MarkDispatched(...)   ─┤                      ├─  TripTopic(wireTopic)                  (FR-3)
          └─ DB span nests UNDER S2                   │   (no MarkDispatched, no broker await)  (FR-4/AC-12b)
             (was orphaned root — C-6 fix)           │
        (no Warning, no breaker — AC-13)             │
                              (both branches wrapped: no throw escapes — NFR-4)

S1 is the original producer span, already ended. S2 is a distinct standalone confirmation span (success or failure) linked back to S1's context — never an enrichment of S1 and never attributed to an unrelated ambient Activity.Current. On success S2 parents the MarkDispatched DB span, replacing today's orphaned (null-parent) Outbox span.

Key Components (roles and responsibilities)

ISupportPublishConfirmation (interfacer / contract). Now knows it must surface the wire topic and the publish span context; its event carries PublishConfirmationResult.
IAmAMessageProducerSync / IAmAMessageProducerAsync (interfacer). Unchanged. Their Send/SendAsync signatures are not modified; the publish span is sourced by the producer from Activity.Current (set to S1 by CreateProducerSpan, BrighterTracer.cs:700), not from a new parameter.
Confirmation-based producers (KafkaMessageProducer, RmqMessageProducer ×2, and opt-in InMemoryMessageProducer) — coordinators between broker and contract. They know the per-message wire topic (they hold message) and the publish-span context (captured from Activity.Current at send time), and do the raising of the enriched event. Kafka additionally knows the failed id (FR-8) and the produce error reason/code (FR-6, via the publisher). InMemoryMessageProducer knows (via a configurable failure hook) whether to confirm success or failure, and does the asynchronous confirm via a worker pump (see below).
OutboxProducerMediator (controller). Decides what a confirmation means on each branch: always request a linked confirmation span; on success keep the "sent" log + mark-dispatched (now parented under the span); on failure warn-log, trip the breaker, leave un-dispatched. Owns both branches and the error-isolation boundary.
IAmAnOutboxCircuitBreaker (service provider). Does the trip via TripTopic(RoutingKey) (src/Paramore.Brighter/CircuitBreaker/IAmAnOutboxCircuitBreaker.cs:44). Already the trip mechanism for the non-confirmation path; the interface is reused unchanged, but the in-tree InMemoryOutboxCircuitBreaker implementation must move its backing Dictionary to a ConcurrentDictionary to be safe under the now-concurrent TripTopic (see "Thread safety (NFR-3)").
BrighterTracer (service provider). Does span creation. Owns the standalone confirmation-span method (used on both branches, with an outcome flag), following the existing CreateProducerSpan/CreateDbSpan/CreateSpan conventions — including setting Activity.Current to the new span so subsequent callback work (the success-branch MarkDispatched DB span) nests under it. (Its ActivitySource is named BrighterSemanticConventions.SourceName = "Paramore.Brighter"; CreateSpan<TRequest> already accepts ActivityLink[]? links at BrighterTracer.cs:106.)

Technology Choices

OpenTelemetry ActivityLink + standalone Activity, created through BrighterTracer using the existing ActivitySource. CreateSpan<TRequest> and CreateBatchSpan<TRequest> already accept and pass an ActivityLink[] to ActivitySource.StartActivity(..., links: links, ...) (BrighterTracer.cs:106, 132), and LinkSpans already constructs new ActivityLink(hs.Value.Context) — so a confirmation span carrying new ActivityLink(ctx) matches an established pattern. Tag conventions reuse BrighterSemanticConventions: MessageId = "messaging.message.id", MessagingDestination = "messaging.destination.name", MessagingOperationType, and an outcome marker — ErrorType = "error.type" is set on the failure branch to mark the publish-confirmation failure, and omitted (or a success status set) on the success branch. Span kind: Producer, consistent with CreateProducerSpan.
Why a standalone span, not span enrichment (C-2). The original producer span has already ended before the confirmation arrives (on either branch), so there is no live span to enrich; reopening an ended span or attributing the confirmation to whatever Activity.Current happens to be active on the callback thread would be incorrect. A short-lived linked span is the OTel-idiomatic representation of an out-of-band signal correlated to a prior operation — and using it on the success branch too replaces the current orphaned (root-parented) MarkDispatched span with one correctly correlated to its publish.

Implementation Approach

No producer Send signature change. The IAmAMessageProducerSync/Async Send/SendWithDelay(Async) signatures are left exactly as they are. The publish span is sourced by the producer from Activity.Current (set to S1 by CreateProducerSpan; see "Capturing the context at dispatch" below), so the ~10 non-confirmation gateways and all existing call sites are untouched. (Adding a RequestContext?/ActivityContext? parameter to the Send methods was considered and rejected — see Alternatives Considered.)

OnMessagePublished contract — before/after (src/Paramore.Brighter/ISupportPublishConfirmation.cs:42):

// before
event Action<bool, string> OnMessagePublished;
// after
event Action<PublishConfirmationResult> OnMessagePublished;   // record defined above

Mediator callback logic (rewritten in both ConfigureAsyncPublisherCallbackMaybe :737 and ConfigurePublisherCallbackMaybe :764). The confirmation span S2 is created first, on every invocation, and becomes the ambient parent for the branch-specific work:

var topic = result.Topic;                               // wire topic; may be null (FR-5)
var id    = result.MessageId;                           // may be empty -> "unknown" marker (FR-5)
var links = result.PublishSpanContext is { } ctx        // FR-2 / AC-2 vs AC-2b
                ? new[] { new ActivityLink(ctx) }
                : null;
using var s2 = _tracer?.CreateConfirmationSpan(id, topic, result.Success, links);  // S2; start+stop in-callback (NFR-2)
                                                        // CreateConfirmationSpan sets Activity.Current = S2 (per CreateDbSpan/CreateProducerSpan convention)
try {
    if (result.Success) {                               // SUCCESS branch — delivery semantics frozen (OOS-4)
        Log.<Info>("sent", id);                         // existing "sent" log, unchanged
        MarkDispatched(id, ctxWithSpan: S2, ...);       // explicitly parent its DB span to S2 (orphan fix, C-6)
                                                        // no Warning, no TripTopic (AC-13)
    }
    else {                                              // FAILURE branch
        Log.<Warning>(id, topic);                       // FR-1/NFR-1
        TripTopic(topic);                               // FR-3 via existing :1168 (no-op if null/empty)
                                                        // no MarkDispatched, no broker/Outbox await (FR-4/AC-12b)
    }
}
catch (Exception ex) { Log.<observability failed, swallowed>(ex); }    // NFR-4 / AC-14 (both branches)

S2 is created inside the callback and disposed deterministically before the callback returns (NFR-2). On the failure branch it is started and stopped with no intervening await; on the success branch it remains the ambient Activity.Current across the single pre-existing MarkDispatched/MarkDispatchedAsync await and is disposed immediately after (a bounded hold across one known Outbox await, not an open-ended span). On success the only new behaviour is S2 + the re-parenting of the existing MarkDispatched DB span; the "sent" log and mark-dispatched themselves are untouched (OOS-4, AC-13). On failure the breaker trip is synchronous against in-memory state, with no MarkDispatched and no awaited broker/Outbox call (AC-12b).

Re-parent the MarkDispatched DB span EXPLICITLY — do not rely on the null-parent fallback. MarkDispatched/MarkDispatchedAsync create the DB span via BrighterTracer.CreateDbSpan(info, parentActivity, options) (BrighterTracer.cs:490), which parents from the explicitly-passed parentActivity — parentId = parentActivity?.Id (:496) — and that argument is the success branch's requestContext.Span (read synchronously at MarkDispatched entry, before any await; e.g. RelationDatabaseOutbox.AddAsync builds the span at the top before WriteToStoreAsync). Today the DB span is orphaned because that requestContext.Span is null on the callback thread (C-6). The orphan-fix is therefore to set the confirmation span S2 as that parent: on the success branch, give MarkDispatched(Async) a RequestContext whose Span == S2 so CreateDbSpan parents the DB span to S2 by its explicit id. We do not lean on the alternative — leaving requestContext.Span null so StartActivity falls back to the ambient Activity.Current == S2 — because that "works" only by the coincidence that the context is empty (C-6); if anything ever populated requestContext.Span, the explicit parentId would silently win and the orphan-fix would regress with no error. Mechanics that make the explicit set clean and race-free: requestContext.Span is thread-keyed by ManagedThreadId (RequestContext.cs:113-125) and read synchronously on the callback thread, so concurrent same-producer confirmations on different threads each set their own S2 with no cross-talk; and because the setter ignores null (cannot unset) and the construction-time requestContext is shared/reused, set Span = S2 on a per-callback copy (requestContext.CreateCopy()) rather than mutating the shared construction-time context. Activity.Current = S2 (set by CreateConfirmationSpan) is retained for ambient hygiene but is no longer the load-bearing mechanism. AC-13's test asserts the DB span's parent id equals S2's id (guaranteed by construction, not by the empty-context coincidence).

Capturing the context at dispatch (C-7). When the mediator creates the producer span S1, CreateProducerSpan sets Activity.Current = S1 (BrighterTracer.cs:700) before Send/SendAsync is invoked. The producer reads Activity.Current?.Context — the value of S1's ActivityContext (a struct, valid after the span ends) — as the first action inside Send/SendAsync, before any await, and stashes that value alongside the per-message confirmation state. This satisfies AC-2's requirement that the link's ActivityContext equal S1's context.

Three correctness points this design depends on:

The source is Activity.Current, which is race-free; NOT the shared producer.Span field, and NOT requestContext.Span. producer.Span (IAmAMessageProducer.Span, IAmAMessageProducer.cs:41) is a single mutable property on a shared, reused producer instance (one per topic in ProducerRegistry), and dispatch is not serialized per producer. Two threads dispatching to the same topic concurrently would race on producer.Span — thread A could read thread B's span and link the failure to the wrong publish span, which AC-2 forbids (FR-2 tolerates a missing link, AC-2b, but not a wrong one; NFR-3/AC-10 require correct behaviour under concurrent same-topic failures). Activity.Current avoids this because it is AsyncLocal — each dispatch thread sees its own value, set to S1 by CreateProducerSpan on that thread. requestContext.Span is also wrong: it holds the parent/ambient span (:812, restored to parentSpan in the finally at :1007), never the producer span.
The value must be captured during the call, never at callback time. S1 is ended by EndSpans in the finally (and Activity.Current is reset — the sync path sets Activity.Current = parentSpan at :857 and nulls producer.Span at :858; the async path restores requestContext.Span at :1007 and relies on the next iteration's overwrite rather than nulling producer.Span). By the time the asynchronous confirmation fires, neither Activity.Current nor producer.Span still holds S1. Capturing the immutable ActivityContext value synchronously at the top of Send — the only window where Activity.Current == S1 — and carrying it forward is therefore mandatory; the captured value is immutable and thread-safe to carry across the confirmation thread (NFR-3).
Activity.Current is not perturbed before the capture. The OutboxProducer resilience pipeline that wraps the send (executed by ExecuteWithResiliencePipeline/Async, OutboxProducerMediator.cs:1111-1160) is the Brighter default pipeline defined in ResiliencePipelineRegistryExtensions.AddBrighterDefault (src/Paramore.Brighter/Extensions/ResiliencePipelineRegistryExtensions.cs:57-67) — a bare AddRetry with no telemetry/activity strategy, and Polly starts no Activity absent its OTel telemetry middleware — so Activity.Current remains S1 from CreateProducerSpan through to the first line of Send/SendAsync. (The pipeline is registry-keyed and user-overridable; this holds for the Brighter default.) The producer MUST read it before doing any work of its own that could start a child activity or await.

When Activity.Current is null (tracing disabled, or no listener so StartActivity returned null), the captured context is default/absent and FR-2 degrades to no link (AC-2b), while FR-1/FR-3 are unaffected (the wire topic comes from message, always present).

Kafka FR-6 (src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessagePublisher.cs:52). The catch currently binds no variable:

// before
catch (ProduceException<string, byte[]>)
// after
catch (ProduceException<string, byte[]> pe)
{
    Log.<Warning>(pe.Error.Reason, pe.Error.Code);   // FR-6, Warning level (NFR-1)
    ... // existing synthetic NotPersisted DeliveryResult + deliveryReport(...) unchanged (FR-7)
}

The synthetic result still writes MESSAGE_ID to deliveryResult.Headers and still invokes the callback (FR-7). This is the correctly-typed catch; the wrong-typed <string, string> catches at KafkaMessageProducer.cs:262, 336 are OOS-2 and left untouched.

Kafka FR-8 + FR-2 (data path). Unlike RMQ, Kafka needs no correlation map: its confirmation is delivered through the per-produce delivery-report closure created inside SendWithDelay/SendWithDelayAsync (KafkaMessageProducer.cs:260 sync, :333 async — today report => PublishResults(report.Status, report.Headers); note Send/SendAsync are thin delegators to these :200, :215). That closure is constructed fresh on each send within the same method that holds the per-message message local — but the current lambda body references only report, so it does not yet capture message. The mechanism is therefore to rewrite the closure body to close over message (and the captured ActivityContext local), at which point the wire topic (message.Header.Topic) and the message id (message.Id) become reachable at confirmation time with no correlation map. This is the Kafka structural analogue of RMQ's PendingConfirmation record — the difference being Kafka's confirmation runs from a closure the producer still holds at confirm time (so the data is captured by closing over it), whereas RMQ's arrives later on a separate delivery-tag handler and must be looked up. The captured publish context rides the same rewritten closure: capture Activity.Current?.Context into a send-method local and let the closure close over it.

Concretely, PublishResults's signature must widen from (PersistenceStatus status, Headers headers) to also receive the captured (RoutingKey topic, ActivityContext? publishContext) (or, equivalently, the closure builds the PublishConfirmationResult inline and hands it to a thin raise helper) — this signature change is the listable Kafka producer-side code change, parallel to RMQ's map-value-type change. For the id, FR-8 still holds: on the NotPersisted fall-through (which today hardcodes OnMessagePublished?.Invoke(false, string.Empty)), read MESSAGE_ID from the same report-level headers collection the success branch uses — not result.Message.Headers, which the publisher leaves empty (KafkaMessagePublisher.cs:58-60) — falling back to empty when genuinely absent (AC-8). (The closure also makes message.Id available as an equivalent source; whichever is used, the id, wire topic, and captured context together populate the enriched PublishConfirmationResult.)

⚠️ Do not source the wire topic from the delivery report. DeliveryReport/DeliveryResult expose a .Topic, but it is not threaded into PublishResults today, and — critically — on the async failure path the publisher fabricates a synthetic DeliveryResult (KafkaMessagePublisher.cs:56-61) that sets only Status, Message, and Headers, never .Topic. So report.Topic is empty on precisely the NotPersisted case this spec exists to observe. The captured message.Header.Topic from the closure is the only reliable source; reaching for the report's topic would silently degrade FR-1/FR-3's topic and FR-2's link on the Kafka failure branch.

Capture-ordering (Kafka). Capture Activity.Current?.Context at the top of SendWithDelay/SendWithDelayAsync, before the _publisher.PublishMessage(Async) call. Kafka's path is less fraught than RMQ's — the non-delay branch has no pre-publish await/child-activity between method entry and the closure construction (WriteProducerEvent writes to the existing Span and does not perturb Activity.Current) — but the capture should still be stated as top-of-method to honour the shared "before any await/child activity" invariant. The callback continues to be marshalled via Task.Run (:373, :381), preserving NFR-2's broker-thread isolation.

RMQ FR-9 + FR-2. The id is already supplied on nack: OnMessagePublished?.Invoke(false, messageId) (RmqMessageProducer.cs:480, async; the RMQ.Sync project is analogous), with a pre-existing Debug log FailedToPublishMessageAsync (:481, :512-513) that may remain (it is not a Warning, so AC-1/AC-9's "exactly one Warning" still holds). For the id, RMQ stays verify-only. RMQ is still not verify-only for FR-2 — but the change is confined to the raise site, not the Send signature: inside its Send/SendAsync the RMQ producer captures Activity.Current?.Context and message.Header.Topic, carries them through its delivery-tag → message tracking, and surfaces them in the enriched PublishConfirmationResult raised at :480. This is not entirely free: the existing _pendingConfirmations map is Dictionary<ulong, string> (deliveryTag → messageId; RmqMessageProducer.cs:60, ConcurrentDictionary in RMQ.Sync). Rather than smear a loose tuple of fields across the tracking code, the map's value type becomes a named record holding exactly the metadata the failure callback needs to build the result:

private readonly Dictionary<ulong, PendingConfirmation> _pendingConfirmations = new();

// internal/private to the RMQ producer (NOT a public API type like PublishConfirmationResult)
internal readonly record struct PendingConfirmation(Id MessageId, RoutingKey Topic, ActivityContext? Context);

AddPendingConfirmation stashes the message id, the wire topic (message.Header.Topic) and the Activity.Current?.Context captured at the top of Send/SendAsync; OnPublishFailed looks the entry up by delivery tag and builds the PublishConfirmationResult from it. This keeps the delivery-tag map focused and intention-revealing instead of polluted with positional fields — and the map is genuinely the right home for this metadata, because it is the only bridge between RabbitMQ's delivery-tag-based, asynchronous, batched-confirm protocol and Brighter's message identity + telemetry context. PendingConfirmation is internal to the producer, so (unlike the public PublishConfirmationResult) it carries no public-API / XML-doc / versioning obligation. The change touches AddPendingConfirmation (:397), RemovePendingConfirmations/RemoveConfirmationsLocked (:419, lookup at :447) and OnPublishFailed (:476); sync analogues at :153, :233. Both RMQ projects change at the raise (and in the map value type), but neither changes its Send signature. Capture-ordering caveat: in SendWithDelayAsync the existing body awaits EnsureBrokerAsync (:167) before it reaches WriteProducerEvent/AddPendingConfirmation, so the Activity.Current capture must be placed at the very top of the method — above BeginSend()/the broker-ensure await — not co-located with the per-message tracking, to honour the "before any await" invariant.

Confirmation-capable in-memory producer (PO-authorised scope expansion). InMemoryMessageProducer (src/Paramore.Brighter/InMemoryMessageProducer.cs) becomes a third ISupportPublishConfirmation implementer so the whole failed-delivery flow is exercisable in-process with a production provider, not a test-only fake. Per ADR 0012, Brighter's in-memory providers are full-featured real implementations — used for fast app-level tests, and also as genuine runtime providers where simplicity outweighs reliability (in-memory bus + consumers in a modular monolith, early development before persistent storage). Leaving the in-memory producer unable to report publish confirmations would make it lag the very abstraction this spec enriches; closing that gap keeps it a first-class provider. Design:

Opt-in switch (property injection, default off). Today the producer writes to the InternalBus and synchronously fires (true, message.Id) inline at all three raise sites (single SendAsync :100, batch SendAsync :123, sync Send :145). A new init-only property bool UseAsyncPublishConfirmation { get; init; } (default false) gates the pump — false preserves today's synchronous write + (true, id) exactly; true routes publish and confirm through the pump. An init-only property (not a ctor arg) gives the class's property-injection style (set via object initializer / DI alongside Publication, Span, Scheduler) and keeps the (bus, publication?, instrumentationOptions) ctor and every existing call site unchanged, while init (rather than set) enforces the "configure before first send" contract at the type level: flipping the switch after the pump's channel and worker are live would leave them in an inconsistent state, so the type forbids it rather than relying on a documented convention. (Updated 2026-06-14 from the original { get; set; } proposal; the unsafe mid-flight-mutation case was not anticipated when set was first written.)
Failure-injection hook. An init-only property Func<Message, bool>? PublishFailurePredicate { get; init; } (default null = never fail), init for the same configure-before-first-send reason as the switch above. The worker invokes it per work-item: true ⇒ treat as a failed publish — do not write the bus, raise (Success: false, …) (message stays not-delivered, consistent with FR-4); null/false ⇒ write the bus, raise (Success: true, …). This drives the false branch end-to-end in-process (FR-1 Warning, FR-2 link, FR-3 TripTopic, FR-5 empty id; AC-5) and, on the default success path, the symmetric success-branch span (FR-2, AC-13).
Fire-and-forget publish + single-worker pump, concurrent raise. When on, Send/SendAsync do not write to the bus inline. Each captures the publish context (below), enqueues a work-item onto an unbounded channel (Channel.CreateUnbounded<T>(new() { SingleReader = true, SingleWriter = false }) — many threads may send concurrently, one worker drains), and returns immediately (fire-and-forget) — the caller never waits on the publish. A single long-running worker Task (lazily started on first enqueue, draining via ChannelReader.ReadAllAsync) processes items in FIFO enqueue order. Because SingleWriter = false lets concurrent Send/SendAsync callers race to be the first enqueuer, the worker MUST be started exactly once — via Interlocked.CompareExchange on a backing Task? field (or LazyInitializer.EnsureInitialized) — so two concurrent first-enqueuers cannot each spin up a reader. SingleReader = true is a caller-asserted contract the channel does not enforce; two concurrent ReadAllAsync readers would break both the FIFO/deterministic-produce-order guarantee and the drain bookkeeping, so the single-start guard is load-bearing, not an optimisation. The worker then, per item: write the message to the InternalBus (unless failure-injected), build the PublishConfirmationResult, then dispatch the raise of OnMessagePublished via Task.Run so confirmation callbacks run concurrently — deliberately mirroring Kafka's delivery-report marshalling (KafkaMessageProducer.cs:373, 381). This faithfully emulates a real confirmation-based producer — Send hands off and returns before acknowledgement; the ack arrives later on another, possibly-overlapping thread — rather than a synchronous publish with a faked-async confirmation. The batch SendAsync(IAmAMessageBatch) path (:112) enqueues one work-item per message. The single serial reader keeps bus-write (produce) order deterministic; only the confirmation raise is concurrent. An unbounded channel can grow if the worker stalls — acceptable for a test/local-dev affordance.
Concurrency scope (NFR-3 / AC-10 — exercised in-memory). Because the worker dispatches each raise via Task.Run, a single producer's same-topic confirmations can overlap, so AC-10's same-topic concurrent callback path (concurrent OnMessagePublished(false, …) for one topic → concurrent mediator TripTopic + Warning logs, no race-induced loss) becomes reachable in-process — the same best-effort model real Kafka uses (Task.Run only queues to the thread pool; it does not guarantee simultaneity). Caveat for the AC-10 test: because the raise bodies are short (log + TripTopic over in-memory state), the pool may run them effectively serially, so a naive test could pass without ever overlapping — giving false confidence. The AC-10 test MUST therefore force overlap with a synchronization gate (e.g. a Barrier/ManualResetEventSlim released inside the injected callback or PublishFailurePredicate) rather than relying on the scheduler to interleave the raises. (RMQ's broker handler raises serially per channel; the pump picks the Kafka concurrency model precisely because it is the one that exercises NFR-3/AC-10.) The trade-off is that cross-message confirmation ordering is non-deterministic — but no AC asserts confirmation order; single-message ACs (AC-2/AC-2b/AC-13) await exactly one confirmation and are unaffected, and AC-10 asserts an order-independent end state. Bus-write order stays deterministic (previous bullet).
Same capture invariant. Capture Activity.Current?.Context synchronously inside Send/SendAsync, before enqueuing (the channel hand-off is the boundary past which Activity.Current is no longer S1), and carry it on the work-item — identical to the Kafka/RMQ invariant — so FR-2/AC-2/AC-2b are exercised in-memory. The work-item is the in-memory analog of RMQ's PendingConfirmation, carrying the Message to publish plus Id, RoutingKey topic, and ActivityContext?.
Delayed sends. The pump engages only on the immediate Send/SendAsync path. SendWithDelay/SendWithDelayAsync with a real delay route through the configured Scheduler (:72), which re-enters Send/SendAsync when the delay elapses — the pump then engages normally and captures the context current at that later send (correct for a delayed publish, whose publish span is created then). Zero/null delay delegates straight to Send/SendAsync, so it too uses the pump.
Lifecycle / draining. The worker + channel are owned by the producer. Because the raise is dispatched via Task.Run, draining is two-stage: Dispose/DisposeAsync complete the channel writer, await the worker (which finishes dequeuing, writing the bus, and spawning the raise tasks), then await the in-flight raise tasks before returning. Two ordering rules make this drain correct, and both MUST be honoured:
- (i) Count before spawn. The worker increments the outstanding-raise count (Interlocked.Increment) immediately before each Task.Run spawn, and the raise decrements it in a finally. The increment in the worker (not inside the Task.Run body) is what makes "await the worker" a guarantee that every raise has already been counted; placing it inside the spawned task would let await worker return with a spawned-but-uncounted raise still pending → a lost confirmation on shutdown (the exact failure this drain exists to prevent).
- (ii) Re-check the already-zero case. The disposer must not wait on a transition that may never re-fire. In the common shutdown case the worker has already drained and every raise has already decremented to zero before the writer was completed, so a TaskCompletionSource keyed purely on "count transitions to zero after writer-complete" would block forever (the transition already happened and won't repeat). After writer.Complete() + await worker, the disposer therefore re-evaluates count == 0 directly and completes (or skips) the TCS if already quiescent; otherwise it awaits the TCS, which the final raise's decrement completes once it observes both zero and writer-completed (bounded wait). Simpler equivalent (preferred): drop the TCS entirely — collect the raise Task handles into a thread-safe bag and await Task.WhenAll(bag) after awaiting the worker; this is immune to both the increment-ordering and the quiescent-transition hazards.
This guarantees no confirmation is silently dropped on shutdown. If the switch was never turned on, no worker is ever started and dispose stays the no-op it is today (:82, :87). Error isolation (NFR-4) wraps each raise so one throwing callback still runs its decrement finally and cannot kill the worker or strand the drain. OnMessagePublished adopts the enriched Action<PublishConfirmationResult> delegate, replacing the Action<bool, Id> (:77, no external subscribers, so no migration cost).

Degradation path. PublishConfirmationResult.PublishSpanContext == null ⇒ no ActivityLink; the standalone confirmation span is still emitted (unconditional, on both branches — FR-2 wording; AC-2b), and on the failure branch the Warning log and breaker trip are likewise unconditional. MessageId empty ⇒ logged as an explicit "unknown" marker; breaker trips on the wire topic regardless (FR-5; AC-5/AC-8). Wire topic null/empty ⇒ TripTopic's guard (:1170) makes the trip a safe no-op.

Thread safety (NFR-3). Callbacks may run concurrently and off the dispatch thread. The breaker (IAmAnOutboxCircuitBreaker implementation) must be safe under concurrent TripTopic. Correction (2026-06-12): the in-tree InMemoryOutboxCircuitBreaker is not currently safe — _trippedTopics is a plain unlocked Dictionary<RoutingKey, int> (src/Paramore.Brighter/CircuitBreaker/InMemoryOutboxCircuitBreaker.cs:42) written by TripTopic via indexer (:70-71) and read by TrippedTopics (:47). Because confirmation failures now reach TripTopic concurrently (the InMemory pump raises via Task.Run; gateway callbacks are concurrent per NFR-3), the backing store must change to ConcurrentDictionary<RoutingKey, int> (indexer-set stays atomic; CoolDown's Remove becomes TryRemove). This is a small behavior-preserving structural change tracked as a /tidy-first task in the task breakdown; the concurrent-trip behavior is pinned by AC-10. Logging via the source-generated ILogger is thread-safe; ActivitySource.StartActivity is thread-safe. We do not read the thread-keyed RequestContext.Span on the callback thread — only the immutable captured ActivityContext value — so there is no cross-thread state hazard from the context flow.

Error isolation (NFR-4 / AC-14). The observability block (log + span) is wrapped so that an exception there is caught and logged but does not propagate out of the callback, and the breaker trip (FR-3) is ordered/guarded to still run. The producer's invoking/thread-pool thread is never destabilised.

Tidy-first note (C-7 "Enabled cleanup" — NOT enabled by this revision). C-7 anticipated that flowing a RequestContext into the producer might make the empty construction-time RequestContext captured in ConfigureCallbacks (:162) removable. Because the revised mechanism sources the span from Activity.Current and does not flow a RequestContext, that cleanup is not enabled here: the construction-time RequestContext stays, still consumed by the success-path MarkDispatched/MarkDispatchedAsync (:749, :776, C-6). No tidy-first change is in scope for this work.

Bulk-dispatch interaction (Correction 2026-06-13). Making InMemoryMessageProducer a confirmation producer makes it the first producer that is both IAmABulkMessageProducerAsync and ISupportPublishConfirmation (Kafka/RMQ are confirmation-capable but not bulk). The bulk sweeper path OutboxProducerMediator.BulkDispatchAsync guarded its bulk-send branch with producer is IAmABulkMessageProducerAsync bulkMessageProducer and not ISupportPublishConfirmation (added in PR #3677, Aug 2025) — an assumption that a bulk producer is never a confirmation producer — and its else simply throws "No async bulk message producer defined.". With InMemory now satisfying both interfaces, the bulk sweeper path (e.g. useBulk: true clears under the circuit breaker) threw. Fix: drop the and not ISupportPublishConfirmation clause so bulk producers take the bulk-send branch regardless of confirmation support; the branch's existing inner guard — if (producer is not ISupportPublishConfirmation && sent) — already defers MarkDispatched to the confirmation callback for confirmation producers (it only marks dispatched inline for non-confirmation ones), exactly mirroring how the non-bulk DispatchAsync path already treats confirmation producers. The !sent ⇒ TripTopic(batch.RoutingKey) behaviour is unchanged, and the else throw remains correct for genuinely non-bulk producers. This is a small production dispatch change beyond the original Kafka+RMQ scope, surfaced by the InMemory confirmation-provider expansion; regression-checked green across InMemory.Tests (144/144 net9, sweeper 4/4 net10) and Core.Tests (759 passed net9).

Consequences

Positive

Confirmation failures become observable for both Kafka and RMQ: a Warning log (FR-1), a standalone linked OTel span navigable back to the originating publish (FR-2), and a circuit-breaker trip (FR-3) — closing all four issue gaps with one shared mediator change plus targeted gateway changes.
Symmetric, correctly-correlated confirmation tracing — and a latent defect fixed. The confirmation span is emitted on both branches, so a successful publish→confirm→dispatch is now a connected trace. This also repairs a pre-existing defect: today the success branch's MarkDispatched Outbox span is parented to the empty construction-time RequestContext (C-6) and surfaces as a disconnected root span; under this change it nests beneath the confirmation span that links to the publish. Framed honestly, the success-branch work is a cleanup of how the callback reports, not a feature add — delivery semantics ("sent" log + mark-dispatched) are untouched.
Exact behavioural parity with the non-confirmation !sent path: same TripTopic, same wire-topic argument, including reply/rewritten-topic messages (NFR-6, AC-3/AC-3b).
Kafka's previously-swallowed produce reason/code is logged (FR-6) and the failed id is no longer discarded (FR-8).
Safe-retry semantics preserved: message stays un-dispatched, nothing bubbles, Sweeper still retries (FR-4, C-1).
The PublishConfirmationResult record is a clean RDD information holder, future-extensible without further signature breaks, and self-documenting at call sites.
Minimal blast radius for the contract. Sourcing the span from Activity.Current means no Send-signature change, so the ~10 non-confirmation gateways and every existing Send/SendAsync call site are untouched. Only the confirmation producers (Kafka, RMQ, and now opt-in InMemory) adopt the enriched delegate.
In-process testability AND a full-featured runtime provider. Making InMemoryMessageProducer confirmation-capable lets the full failed-delivery flow (FR-1/2/3/5, the false path, the cross-thread callback, and the FR-2 ActivityLink) be tested fast and in-process without Kafka/RMQ infrastructure — using a real provider rather than a test-only fake. Per ADR 0012 this same capability serves real runtime use (modular monolith, early development before persistent storage), so the in-memory producer reports confirmations like the broker-backed ones instead of lagging the abstraction. Also tidies the previously-orphaned Action<bool, Id> event.
The in-memory confirmation-provider decision proved its worth empirically (2026-06-13). Bringing InMemoryMessageProducer to confirmation parity — rather than testing the mediator with a test-only fake that mimics only the parts we already understood — flushed out a latent production defect we would otherwise not have caught: the bulk-dispatch path assumed no producer is both bulk and confirmation-capable and would throw for any future producer that is (see "Bulk-dispatch interaction"). Kafka/RMQ are not bulk producers, so neither the existing broker tests nor a hand-rolled fake would have exercised that combination; only a real, full-featured in-memory provider that happens to be both surfaced it. This is a concrete vindication of the ADR 0012 philosophy (full-featured in-memory providers, not fakes) and confirms the PO-authorised choice to use the in-memory producer here.
End users see no API change.

Negative

Source- and binary-breaking for confirmation-producer implementors. Changing OnMessagePublished from Action<bool, string> to Action<PublishConfirmationResult> breaks any in-tree or third-party gateway that implements ISupportPublishConfirmation and raises the event; they must update their raise sites, the mediator's two handlers must be rewritten to the new delegate shape, and precompiled assemblies referencing the old delegate type will fail to bind until recompiled. End users and non-confirmation gateways are unaffected, but this is a real break for the confirmation-gateway-author audience and must be called out in release notes. (This is the only breaking surface — non-confirmation gateways are untouched.)
RMQ is not verify-only for FR-2. The RMQ producers must capture Activity.Current/topic and surface them in the enriched raise, replacing the _pendingConfirmations map value with a named internal PendingConfirmation record; only the id handling (FR-1/3/4) stays verify-only. The change is confined to the raise site + map, not the Send signature.
Coupling to Activity.Current at dispatch and a narrow capture window. The design relies on CreateProducerSpan having set Activity.Current = S1 before Send and on the producer capturing Activity.Current?.Context synchronously as the first action, before any await or any work that could start a child activity. If a future change started an intervening activity between span creation and the Send body (e.g. an activity-emitting resilience strategy on the OutboxProducer pipeline), or a producer read Activity.Current after an await, the captured context would be wrong or null. This invariant must be documented at both ends and pinned by tests (AC-2). Note this is a narrow same-thread ordering invariant, not a cross-thread race — Activity.Current is AsyncLocal, so concurrent dispatch to the same producer does not corrupt it (the reason the shared producer.Span field was rejected as the source).
New public type to document and version. PublishConfirmationResult is an additive public type requiring XML docs on the type and all members, and it becomes part of the versioned public API surface.
New obligations on both callback branches. Thread-safety (NFR-3) and error-isolation (NFR-4) become standing requirements on the mediator's callback (now on the success branch too) and the breaker implementation; getting either wrong reintroduces the very instability this change is meant to avoid. Concretely, this surfaced that InMemoryOutboxCircuitBreaker was using an unlocked Dictionary and must move to ConcurrentDictionary (see "Thread safety (NFR-3)") — a real, if small, production change rather than a no-op.
The success branch is no longer untouched. Emitting the confirmation span on success is an always-on cost on the hot path (every confirmed publish, whenever a tracing listener is active), not just the rare failure path — and the success handler now starts/stops a span and re-parents the MarkDispatched DB span. The change is deliberately narrow (delivery semantics frozen — OOS-4 — and the span is short-lived, NFR-2), and it corrects an existing defect rather than inventing behaviour, but it does mean OOS-4 was narrowed (success delivery only) and the success path now needs its own regression coverage (AC-13). Anyone who previously relied on the orphaned MarkDispatched span's shape (e.g. asserting it has no parent) will see it re-parented.
InMemory scope increase. Making InMemoryMessageProducer confirmation-capable adds real production code to a previously-simple component: a config switch, a Channel + single-worker pump that dispatches each raise via Task.Run (for AC-10 concurrency), a two-stage drain on dispose (await the worker, then await the outstanding raise tasks — preferably via Task.WhenAll over the raise handles, or an Interlocked count + TCS with the quiescent-zero re-check), and a failure-injection hook — and NFR-3/NFR-4 now apply to the pump too. The concurrent-raise design is the most error-prone part (drain correctness, no lost/double confirmations) and must be covered by its own tests. This is a PO-authorised expansion beyond the original Kafka+RMQ scope, but it is consistent with ADR 0012 (full-featured in-memory providers, not fakes) rather than scope creep: its payoff is both in-process testability — including the otherwise Kafka-only NFR-3/AC-10 concurrency path — and a confirmation-capable in-memory provider usable at runtime where simplicity outweighs reliability (modular monolith, early development). The added complexity is real and earns its own tests; it is not throwaway test scaffolding. Its OnMessagePublished event changes type (enriched delegate), which is a source/binary break for that event in principle, though there are no in-repo external subscribers.
Deferred bus visibility when the InMemory switch is on. With UseAsyncPublishConfirmation = true, Send/SendAsync return before the message is written to the InternalBus (the worker writes it), so a test that calls Send and then synchronously inspects the bus may not see the message yet. This is the whole point (it emulates fire-and-forget async publishing), but it means such tests must await the confirmation / poll the bus rather than read immediately. The default (false) keeps today's synchronous read-after-Send behaviour, so existing in-process tests are unaffected unless they opt in.
Bulk-dispatch branch condition changed. Because InMemoryMessageProducer is now both bulk and confirmation-capable, BulkDispatchAsync had to stop excluding confirmation producers from its bulk-send branch (drop and not ISupportPublishConfirmation at :907; see "Bulk-dispatch interaction" above). This is a real, if small, production dispatch change rather than a no-op — bulk dispatch of a confirmation producer now defers MarkDispatched to the callback (the inner producer is not ISupportPublishConfirmation && sent guard already handled this) instead of throwing. Behaviour for non-confirmation bulk producers and for genuinely non-bulk producers is unchanged.

Risks and Mitigations

Reading the wrong Kafka header collection (FR-8): reading result.Message.Headers yields empty because the publisher writes only to deliveryResult.Headers (KafkaMessagePublisher.cs:58-60). Mitigation: read the report-level headers the success branch already uses; pin with AC-7/AC-8 tests.
Sourcing the Kafka wire topic from the delivery report instead of the captured message (FR-1/FR-2/FR-3): DeliveryReport/DeliveryResult.Topic is not threaded into PublishResults, and the synthetic NotPersisted DeliveryResult never sets .Topic (KafkaMessagePublisher.cs:56-61), so it is empty on exactly the failure path this spec observes — using it would degrade the breaker-trip topic, the Warning-log topic, and the FR-2 link on the Kafka failure branch. Mitigation: source the wire topic solely from the closure's message.Header.Topic; pin with a Kafka-specific link/topic assertion (the otherwise in-memory-only AC-2/AC-2b/AC-3b coverage does not exercise the Kafka closure).
Capturing the span context at the wrong moment, or from the wrong source, returns the wrong span or null. The shared producer.Span field races under concurrent same-topic dispatch (rejected); requestContext.Span is the parent (rejected). Mitigation: capture Activity.Current?.Context (AsyncLocal, race-free) as the first action inside Send/SendAsync, before any await or child-activity creation, while Activity.Current == S1; carry the immutable ActivityContext value, never re-read it at callback time. Pin with AC-2 (link equals S1), AC-2b (degrades to no link), and AC-10 (concurrent same-topic failures link to the correct respective spans).
Observability throw destabilises the producer thread (AC-14). Mitigation: wrap observability work; prioritise breaker trip + Warning log.
Mis-tripping Publication.Topic instead of the wire topic for reply messages. Mitigation: always use result.Topic sourced from message.Header.Topic; pin with AC-3b.
Span left open across an await (NFR-2). Mitigation: create and dispose the standalone span synchronously inside the callback; add no new awaited broker/Outbox call on the failure path (AC-12/AC-12b).

Alternatives Considered

Flowing the publish context via a new RequestContext? parameter on the Send methods (a heavier way to source the same context). The original requirement sketch (former FR-10) proposed adding an optional RequestContext? parameter to IAmAMessageProducerSync.Send/SendWithDelay and IAmAMessageProducerAsync.SendAsync/SendWithDelayAsync, populated by Dispatch/DispatchAsync, so the producer could surface the publish span to the callback. Genuine appeal: an explicit, type-checked channel for the context, aligned with the RequestContext already flowing through the mediator. Why rejected: (1) it changes the IAmAMessageProducerSync/Async contracts, so all ~10 non-confirmation gateways (InMemoryMessageProducer, AWS SNS/SQS V3+V4, Azure Service Bus, GCP, MQTT, MsSql, Postgres, Redis, RocketMQ) would absorb a parameter they never use, when only Kafka and RMQ need the context; (2) the two async methods (SendAsync/SendWithDelayAsync) already end in CancellationToken cancellationToken = default (IAmAMessageProducerAsync.cs:49,57), and many in-tree callers pass the token positionally (e.g. KafkaMessageConsumer.cs:696,766; OutboxProducerMediator.cs:908,978), so any parameter inserted ahead of it is a compile break — the two sync methods (IAmAMessageProducerSync.cs:45,52) take no token but would still force a signature change across every implementor; (3) it is unnecessary — CreateProducerSpan already sets Activity.Current = S1 before Send is called (BrighterTracer.cs:700), so the producer can self-source the span value with no signature change. The Decision captures Activity.Current?.Context instead, confining all change to the confirmation contract while delivering the identical ActivityLink (FR-2).

(b) messageId → ActivityContext correlation map. Populate a shared map in Dispatch/DispatchAsync keyed by message id, and read it in the callback. Genuine appeal: no contract change, so no all-gateways ripple and no source break. Why rejected: it introduces shared mutable state across the dispatch thread and the gateway-dependent callback threads (NFR-3), requiring its own concurrency control and — critically — a lifecycle/eviction policy. Confirmations are asynchronous and may never arrive (a permanently failed broker), so entries leak unless evicted on a timer or TTL, which races against late-arriving confirmations and can either drop a still-valid context or evict too late. It also still does not solve C-8 cleanly: the wire topic would have to be co-stored in the map, duplicating knowledge the producer already holds in message. It trades a one-time, compile-time-checked contract break for an ongoing runtime correctness hazard.

(c) Header propagation / read trace context back from message headers. Write the publish span's trace context into the message headers (Brighter already round-trips TraceParent), and have the callback reconstruct the ActivityContext from the message. Genuine appeal: leans on an existing W3C-traceparent mechanism; no producer signature change for the context. Why rejected: the callback (per C-6) does not capture the message, so the headers are not reachable there without also plumbing the message to the callback — i.e. it still needs a contract or map change, so it does not actually avoid the cost. It conflates the consumer-side parent context (what traceparent is for) with an internal producer→confirmation link, risking incorrect parenting of downstream consumer spans. And it only addresses the trace context, leaving C-8's wire-topic plumbing unsolved. Mechanism (a) carries both the context and the wire topic through one explicit, type-checked channel.

References

Requirements: specs/0034-failed-delivery-context/requirements.md
Related ADRs: ADR 0012 — Provide In-Memory and Simple Implementations (the basis for making InMemoryMessageProducer a full-featured confirmation provider rather than a test-only fake). This is otherwise the first and only ADR for this spec.
External references:
- OpenTelemetry span links: https://opentelemetry.io/docs/concepts/signals/traces/#span-links
- .NET System.Diagnostics.ActivityLink / ActivityContext: https://learn.microsoft.com/dotnet/api/system.diagnostics.activitylink

Critical Files for Implementation

src/Paramore.Brighter/OutboxProducerMediator.cs — mediator callback handlers (rewritten to the new delegate shape; both branches gain the linked confirmation span; failure branch additionally gains Warning log + breaker trip; success branch keeps "sent" log + mark-dispatched, now nested under the span), TripTopic at :1168; BulkDispatchAsync branch condition at :907 (drop and not ISupportPublishConfirmation so bulk+confirmation producers dispatch and defer MarkDispatched to the callback — see "Bulk-dispatch interaction")
src/Paramore.Brighter/Observability/BrighterTracer.cs — CreateProducerSpan sets Activity.Current = S1 (:700, the capture source); new standalone confirmation-span helper (outcome flag, both branches) using ActivityLink, setting Activity.Current to the new span (per CreateDbSpan/CreateProducerSpan convention) so the success-branch MarkDispatched span nests under it
src/Paramore.Brighter/ISupportPublishConfirmation.cs — OnMessagePublished delegate change to Action<PublishConfirmationResult>
src/Paramore.Brighter/PublishConfirmationResult.cs (new) — the public result record (with XML docs)
IAmAMessageProducerSync/IAmAMessageProducerAsync — no signature change (Activity.Current, not a Send parameter, carries the span)
src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessagePublisher.cs and KafkaMessageProducer.cs — FR-6 logging, FR-8 id propagation, capture Activity.Current?.Context in Send/SendAsync (delivery-report closure), enriched raise
src/Paramore.Brighter.MessagingGateway.RMQ.Async/RmqMessageProducer.cs and src/Paramore.Brighter.MessagingGateway.RMQ.Sync/RmqMessageProducer.cs — capture Activity.Current?.Context/topic, introduce internal PendingConfirmation record as the _pendingConfirmations value, enriched raise (no Send-signature change)
src/Paramore.Brighter/InMemoryMessageProducer.cs — implement ISupportPublishConfirmation with the enriched event (replacing the orphaned Action<bool, Id>); add opt-in switch routing publish+confirm through a Channel/worker pump (fire-and-forget Send: enqueue and return; worker writes the bus then raises the confirmation) + failure-injection hook + dispose draining; capture Activity.Current before enqueue

Navigation Menu

Repository navigation

0063-failed-delivery-context.md

0063-failed-delivery-context.md

0063. Failed Delivery Context for Confirmation-Based Producers

Status

Context

Decision

Choosing the enriched `OnMessagePublished` shape

Architecture Overview

Key Components (roles and responsibilities)

Technology Choices

Implementation Approach

Consequences

Positive

Negative

Risks and Mitigations

Alternatives Considered

References

Critical Files for Implementation

FilesExpand file tree

0063-failed-delivery-context.md

Latest commit

History

0063-failed-delivery-context.md

File metadata and controls

0063. Failed Delivery Context for Confirmation-Based Producers

Status

Context

Decision

Choosing the enriched OnMessagePublished shape

Architecture Overview

Key Components (roles and responsibilities)

Technology Choices

Implementation Approach

Consequences

Positive

Negative

Risks and Mitigations

Alternatives Considered

References

Critical Files for Implementation

Choosing the enriched `OnMessagePublished` shape