Skip to content

requirements.md

Latest commit

 

History

History
175 lines (119 loc) · 35.9 KB

requirements.md

File metadata and controls

175 lines (119 loc) · 35.9 KB
Raw

Requirements

Note: This document captures user requirements and needs. Technical design decisions and implementation details should be documented in an Architecture Decision Record (ADR) in docs/adr/.

Linked Issue: #4179

Glossary

These terms are used with precise meaning throughout this document.

  • Confirmation-based producer: A Brighter message producer that implements ISupportPublishConfirmation and reports publish success/failure asynchronously (after the synchronous Send/SendAsync call has already returned) by raising the OnMessagePublished event. The RabbitMQ (RmqMessageProducer) and Kafka (KafkaMessageProducer) producers are confirmation-based. This is distinct from a non-confirmation producer, whose send either succeeds or throws synchronously and is observed inline via the sent boolean in OutboxProducerMediator.Dispatch (sync) and OutboxProducerMediator.DispatchAsync (async).
  • ISupportPublishConfirmation: The Brighter interface exposing the OnMessagePublished event, whose signature is OnMessagePublished(bool success, string messageId). success == true means the broker acknowledged (acked) the message; success == false means the broker rejected/failed to persist (nacked / NotPersisted / produce error) the message.
  • OnMessagePublished callback: The handler wired up by OutboxProducerMediator (in ConfigureAsyncPublisherCallbackMaybe / ConfigurePublisherCallbackMaybe) onto a confirmation-based producer's event. It is a plain delegate (a function pointer): it executes synchronously on whichever thread raises the event, which is gateway-dependent — Kafka marshals the invocation through Task.Run onto a thread-pool thread (see KafkaMessageProducer.PublishResults), whereas RabbitMQ invokes it directly from its broker ack/nack handler. The callback itself implies no dedicated thread; it is simply not guaranteed to run on the thread that called Dispatch/DispatchAsync.
  • Sweeper (Outbox Sweeper): The Brighter background component that periodically scans the Outbox for messages that were written but not marked dispatched, and re-attempts delivery. A message that fails to publish is left un-dispatched precisely so the Sweeper retries it.
  • Mark dispatched: The Outbox operation that records a message as successfully delivered (so the Sweeper will not retry it). This happens today only on the success == true branch of the callback.
  • Circuit breaker trip (TripTopic): The OutboxProducerMediator runtime action (OutboxProducerMediator.cs:998 calls it on the non-confirmation !sent path) that marks a topic as temporarily un-publishable after a delivery failure, so Brighter stops attempting that topic until the breaker resets. Tripping is per-topic.
  • Standalone linked span: A new, short-lived OpenTelemetry Activity (span) created at callback time to represent the publish-confirmation event (success or failure), carrying an ActivityLink to the original publish span's ActivityContext (when reachable; see C-7), rather than reusing the original (already-ended) producer span. On the success branch it also becomes the parent of the Outbox MarkDispatched span.
  • Wire topic (message.Header.Topic): The routing key the message is actually being sent to. For most publications it equals the producer's configured Publication.Topic, but for reply/RPC messages it is rewritten to a dynamic reply address (see GetProducerLookupTopic, OutboxProducerMediator.cs:786-807). This is the topic the non-confirmation !sent path logs and trips (OutboxProducerMediator.cs:998), and the topic FR-1/FR-3 use (see C-8).
  • Publication.Topic: The static routing key configured on a producer's Publication, declared RoutingKey? Topic { get; set; } (Publication.cs:86) — nullable in type, but non-null at callback time by the ProducerRegistry registration invariant (registration throws ConfigurationException on a null/empty topic). ProducerRegistry keys producers by a composite ProducerKey(Publication.Topic, CloudEvents type) (ProducerRegistry.cs:29, 52), not by topic alone. It is not used as the breaker/log topic, because it diverges from the wire topic for reply messages (see C-8).

Problem Statement

As an operator/SRE running a Brighter application that publishes via confirmation-based producers (RabbitMQ or Kafka), I would like publish failures that are reported asynchronously via OnMessagePublished(success: false, ...) to be logged, traced, and to trip the topic circuit breaker so that I can see and alert on failed deliveries, correlate them to the original publish, and get the same protective circuit-breaker behaviour I already get from non-confirmation producers — while the message safely remains un-dispatched in the Outbox for the Sweeper to retry.

Today the success == false case is silently dropped: nothing is logged, no telemetry is emitted, the circuit breaker is never tripped for confirmation-based producers, and for Kafka the failed message id and the underlying ProduceException reason/code are discarded. Operators have no signal that delivery failed.

Proposed Solution

When a confirmation-based producer reports the outcome of a publish via OnMessagePublished, Brighter will make that outcome observable and (on failure) protective, without changing the existing safe-retry behaviour:

  1. On failure (success == false), log at Warning level (not Error — the message is not lost; it stays in the Outbox for Sweeper retry), including the message id when available.
  2. On every confirmation (success and failure), emit a standalone, short-lived OpenTelemetry span that carries an OpenTelemetry Link to the original publish span (via that span's context/id), so the confirmation is visible in tracing tools and navigable back to the originating publish even though the original producer span has already ended. This requires the original publish span's context to be reachable at callback time (see C-7). This also corrects a pre-existing defect: today the success branch's MarkDispatched Outbox span is emitted orphaned (no parent) — under this change it nests beneath the confirmation span, which links to the publish.
  3. On failure, trip the topic's circuit breaker, so confirmation-based producers behave like non-confirmation producers do on a failed send.

Additionally, for Kafka specifically, the underlying produce error (reason + code) will be logged where it is currently swallowed, and the failed message's id will be propagated through to OnMessagePublished so the log/telemetry carry it (RabbitMQ already supplies the id and only needs verification).

The message is still left un-dispatched; the Sweeper still retries it; produce exceptions are still not bubbled to the caller.

Requirements

Each FR is labelled with its scope:

  • [SHARED] — implemented once in OutboxProducerMediator callback configuration; applies to all ISupportPublishConfirmation producers (Kafka + RabbitMQ, sync + async).
  • [KAFKA] — Kafka producer/publisher-side change.
  • [RMQ-VERIFY] — RabbitMQ already behaves correctly; requirement is verification only, no behaviour change.

The four gaps from the issue map to FRs as follows: no log → FR-1, FR-6; id discarded → FR-5, FR-8; breaker not tripped → FR-3; no OTel signal → FR-2.

Functional Requirements

FR-1 [SHARED] — Log confirmation failure at Warning. When OnMessagePublished fires with success == false, the mediator callback (ConfigureAsyncPublisherCallbackMaybe and ConfigurePublisherCallbackMaybe, both async and sync variants) MUST emit a log entry at Warning level. The log entry MUST include the message id (when non-empty) and the wire topic message.Header.Topic (the same topic the non-confirmation path logs; reachable at callback time per C-8). It MUST NOT be logged at Error or higher. Example: Producer raises OnMessagePublished(false, "abc-123") for a message bound to topic orders. Result: exactly one Warning log line is emitted containing message id abc-123 and topic orders; no Error line is emitted.

FR-2 [SHARED] — Emit a standalone OTel span linked to the original publish span, on both success and failure confirmations. When OnMessagePublished fires — whether success == true or success == false — the mediator callback MUST emit a new, standalone, short-lived OpenTelemetry Activity (S2) representing the publish-confirmation event. This Activity MUST carry an OpenTelemetry Link (ActivityLink) to the original publish span's context (the producer span S1 created during Dispatch/DispatchAsync) when that context is reachable at callback time (C-7); when it is not reachable, the Activity is emitted without the Link (graceful degradation). The Link, when present, lets tracing tools navigate from the confirmation back to the originating publish, even though S1 has already ended. The Activity MUST carry at least: message id (or an explicit "unknown" marker when empty), the wire topic message.Header.Topic (C-8), the outcome (confirmed vs failed), and an indication that this is a publish-confirmation event. Regardless of whether the Link is present, the callback MUST NOT reopen, mutate, or enrich the already-ended S1, and MUST NOT attribute the confirmation to whatever unrelated ambient Activity.Current happens to be active.

This generalises a behaviour that the failure branch lacked entirely and that the success branch did incorrectly: today the success branch's MarkDispatched/MarkDispatchedAsync Outbox call emits its DB span orphaned (parented to the empty construction-time RequestContext whose Span is null — C-6 — so it surfaces as a disconnected root span, unlinked from the publish). Under this FR the confirmation Activity S2 becomes the ambient parent for any work the callback then does, so on the success branch the MarkDispatched DB span nests under S2 (which links to S1) instead of floating free. Only the Link aspect is contingent on the C-7 mechanism; emitting S2 is unconditional. The Warning log (FR-1) and breaker trip (FR-3) remain failure-only — S2 is the only observability addition on the success branch (no Warning, no breaker trip; see AC-13). Failure example: Original publish of abc-123 created producer span S1 (now ended); S1's ActivityContext was made reachable at callback time (C-7; mechanism per ADR 0063). Confirmation failure fires later. Result: a distinct Activity S2 is created (outcome = failed) carrying an ActivityLink to S1's ActivityContext; S1 itself is not mutated. Success example: Confirmation success for abc-123 fires. Result: a distinct Activity S2 (outcome = confirmed) carrying an ActivityLink to S1 is created; the MarkDispatched DB span is emitted as a child of S2 (no longer an orphaned root span). The "sent" Info log and mark-dispatched still occur (AC-13). Degraded example: if no context is reachable (e.g. tracing disabled), S2 is still emitted on either branch but carries no Link.

FR-3 [SHARED] — Trip the topic circuit breaker on failure. When OnMessagePublished fires with success == false, the mediator MUST trip the circuit breaker for that message's wire topic message.Header.Topic — the same topic argument and TripTopic action used by the non-confirmation !sent path (OutboxProducerMediator.cs:998), giving exact parity. For reply/rewritten-topic messages, where the wire topic differs from Publication.Topic, the wire topic is the one tripped (matching the non-confirmation path; not Publication.Topic). The wire topic must be reachable at callback time (C-8). This is a runtime behaviour change. Example: Two messages whose wire topic Header.Topic is payments both nack via OnMessagePublished(false, ...). Result: the payments topic is tripped, identical to what a synchronous !sent failure on payments would have done. Reply-topic example: a reply message whose Header.Topic was rewritten to reply-xyz (≠ its Publication.Topic) trips reply-xyz.

FR-4 [SHARED] — Preserve un-dispatched / no-bubble semantics. On success == false, the mediator MUST NOT mark the message dispatched in the Outbox and MUST NOT throw or bubble any exception to the original caller. The message MUST remain eligible for Sweeper retry. Example: After OnMessagePublished(false, "abc-123"), querying the Outbox shows abc-123 still un-dispatched; the calling code that invoked Post/DepositPost observed no exception.

FR-5 [SHARED] — Handle missing/empty message id gracefully. When OnMessagePublished fires with success == false and an empty or null message id, FR-1 (Warning log), FR-2 (OTel span) and FR-3 (breaker trip — using the wire topic message.Header.Topic per C-8, not the id) MUST still occur. The log/telemetry MUST record the id as an explicit "unknown"/empty marker rather than crashing or omitting the entry. Example: Producer raises OnMessagePublished(false, ""). Result: one Warning log emitted with id shown as empty/unknown, one OTel signal emitted, and the breaker for the associated topic is tripped; no exception is thrown.

FR-6 [KAFKA] — Log the swallowed ProduceException. In KafkaMessagePublisher.PublishMessageAsync (KafkaMessagePublisher.cs:52-62), where a ProduceException<string, byte[]> is currently caught and a synthetic DeliveryResult with Status = NotPersisted is created with NOTHING logged, the code MUST log the produce error including the broker error reason and code. The current catch clause binds no exception variable (catch (ProduceException<string, byte[]>)); satisfying this FR requires binding it (e.g. catch (ProduceException<string, byte[]> pe)) so pe.Error.Reason/pe.Error.Code can be read. (This catch is correctly typed — distinct from the dead, wrong-typed <string, string> catches noted in OOS-2.) Level: Warning (consistent with FR-1; not Error). The synthetic DeliveryResult and the existing non-bubbling flow are otherwise preserved. Example: Confluent client throws ProduceException with Error.Reason = "Broker: Message size too large", Error.Code = MsgSizeTooLarge. Result: a Warning log line is emitted containing both Message size too large and MsgSizeTooLarge; PublishMessageAsync still returns the synthetic NotPersisted result without rethrowing.

FR-7 [KAFKA] — Preserve synthetic NotPersisted callback flow. The Kafka publisher MUST continue to synthesise the DeliveryResult with Status = NotPersisted and route it to the callback (so OnMessagePublished(false, ...) still fires). FR-6 adds logging only; it MUST NOT cause the produce path to start bubbling the exception or to skip invoking the callback. Example: After the ProduceException of FR-6, KafkaMessageProducer.PublishResults still receives the synthetic report and OnMessagePublished still fires for the failed message.

FR-8 [KAFKA] — Propagate failed message id through PublishResults. In KafkaMessageProducer.PublishResults (KafkaMessageProducer.cs:364-384), the NotPersisted branch MUST read the MESSAGE_ID header from the (synthetic) delivery report — which the publisher already populates — and fire OnMessagePublished(false, <messageId>) with that id, instead of the current OnMessagePublished(false, string.Empty). The id MUST be read from the report-level Headers collection passed into PublishResults (the same collection the Persisted branch reads via result.Headers at KafkaMessageProducer.cs:333, 366), NOT from result.Message.Headers — the publisher writes MESSAGE_ID to deliveryResult.Headers while leaving deliveryResult.Message.Headers empty (KafkaMessagePublisher.cs:58-60); reading the wrong collection would yield empty. When the header is genuinely absent, fall back to empty (handled by FR-5). Example: Synthetic NotPersisted report carries MESSAGE_ID = "abc-123". Result: OnMessagePublished(false, "abc-123") is fired (not (false, "")); the id then flows into FR-1/FR-2.

FR-9 [RMQ-VERIFY] — RabbitMQ already supplies the id; verify only. The RabbitMQ producer already invokes OnMessagePublished(false, messageId) with the message id on nack (RmqMessageProducer.cs:480 async; sync analogous). For the id, no RMQ producer-side change is required. Requirement: provide/confirm test coverage that the RMQ nack path carries the id into the shared mediator handling (FR-1/FR-2/FR-3/FR-4). Example: RMQ broker nacks message def-456. Result: OnMessagePublished(false, "def-456") fires; the shared mediator path produces the Warning log (with def-456), OTel signal, and breaker trip for the RMQ topic.

Caveat (see C-7): the verify-only classification covers the id-handling path (FR-1/FR-3/FR-4) only. Whether RMQ needs a producer-side change to make FR-2's span Link reachable depends on the C-7 mechanism the ADR selects; under some mechanisms RMQ is not verify-only for FR-2. Pre-existing RMQ Debug log: the RMQ async producer already logs the nack itself at Debug via Log.FailedToPublishMessageAsync (RmqMessageProducer.cs:480-481, 512-513). This Debug line MAY remain; the new shared Warning (FR-1) is the operator-facing signal. FR-1's "exactly one Warning" and AC-1/AC-9 count Warning-level lines specifically — a co-existing Debug line is acceptable and is not a duplicate Warning. Removing/downgrading the producer-side Debug log is out of scope.

Non-functional Requirements

  • NFR-1 (Log level). Confirmation-failure logging MUST be at Warning, never Error/Critical. A failed-but-retained message is an expected, recoverable condition (Sweeper will retry), not data loss.
  • NFR-2 (Low overhead, non-blocking — concrete bar). The callback's confirmation handling MUST: (i) perform no synchronous network I/O on the callback thread for logging/telemetry; (ii) create the standalone OTel span within the callback and dispose it deterministically before the callback returns (short-lived, never abandoned across the lifetime of the callback) — on both branches, since the span (FR-2) is now emitted on both. On the failure branch the span is started and stopped with no intervening await. On the success branch the span remains live across the single pre-existing MarkDispatched/MarkDispatchedAsync await — because the DB span is parented to it (the orphan-fix, FR-2/AC-13; the parenting is explicit, see ADR 0063) — and is disposed immediately after, before the callback returns; this is a bounded hold across one known Outbox await, not an open-ended span left dangling. (iii) On the failure branch, introduce no new await on broker or Outbox operations beyond the existing breaker trip and log. (The success branch legitimately awaits the pre-existing MarkDispatched/MarkDispatchedAsync; FR-2 adds no further broker/Outbox await there.) No other latency bar is implied (the prior vague "measurably delay"/"cheap" wording is replaced by (i)–(iii)). Note: Kafka already isolates the broker thread by raising the callback via Task.Run (KafkaMessageProducer.cs:373, 381).
  • NFR-3 (Thread safety). Confirmation callbacks may be invoked concurrently (multiple messages in flight) and on a thread other than the one that dispatched the message — which thread is gateway-dependent (Kafka marshals via Task.Run onto a thread-pool thread; RabbitMQ invokes from its broker confirmation handler). The callback-side work (FR-1/2/3) MUST therefore be safe under concurrent invocation; access to the circuit breaker and any shared logging/telemetry state MUST be thread-safe. This requirement stands because of possible concurrency, not because the callback is inherently a "background thread" (it is a synchronous delegate running on the invoking thread).
  • NFR-4 (No swallowed errors in the observability code). Failure within the observability handling itself (e.g. telemetry emission) MUST NOT throw out of the callback in a way that destabilises the producer's background thread; it must degrade gracefully (the breaker trip and Warning log are the priority).
  • NFR-5 (Always-on). This observability/breaker behaviour is always on; no new public configuration toggle is introduced (see Out of Scope).
  • NFR-6 (Consistency with existing path). The breaker trip on confirmation failure MUST use the same circuit-breaker mechanism/TripTopic semantics as the existing non-confirmation !sent path — including tripping the same per-message wire topic message.Header.Topic (OutboxProducerMediator.cs:998), not Publication.Topic (which diverges from the wire topic for reply/rewritten topics). Behaviour is therefore identical across producer types.

Constraints and Assumptions

  • C-1. Brighter deliberately does NOT bubble produce exceptions to the caller; failed messages are left un-dispatched for the Sweeper to retry. This spec makes failure observable and trips the breaker; it does not change retry/non-bubbling semantics.
  • C-2. The OnMessagePublished confirmation callback fires after Dispatch/DispatchAsync has already ended its producer spans (broker confirmation is asynchronous relative to the send call returning), so there is typically no live producer span to enrich — this is the rationale for the standalone-span + Link decision (FR-2), independent of which thread raises the callback.
  • C-3. The Kafka publisher already places MESSAGE_ID into the synthetic NotPersisted report's headers, so FR-8 is a read-and-forward, not a new header write.
  • C-4. The RabbitMQ producer already carries the message id on nack; only verification (FR-9) is needed, no RMQ producer-side code change.
  • C-5. "Confirmation-based producer" = any type implementing ISupportPublishConfirmation (currently Kafka and RabbitMQ, sync + async variants). The shared mediator fix (FR-1/2/3/4/5) is written once and applies to all of them.
  • C-6 (Fact, verified — corrects an earlier assumption). The confirmation callback is wired once per producer in OutboxProducerMediator's constructor via ConfigureCallbacks (OutboxProducerMediator.cs:162), iterating _producerRegistry.Producers — it is not re-registered per outgoing message nor per Dispatch/DispatchAsync. The closure captures a single construction-time RequestContext (from requestContextFactory.Create()) that is empty of per-publish data: it carries neither the per-message topic nor the per-publish producer span. Therefore the topic and the original trace context for FR-1/FR-2/FR-3 cannot be taken from that captured RequestContext. The captured producer yields only the static Publication.Topic, which the wire-topic decision rejects (C-8); the per-message wire topic message.Header.Topic is not reachable from the captured closure state (the closure does not capture the per-message message) and must be plumbed to the callback (see C-8), alongside the original trace context, which is likewise not currently available at callback time (see C-7). (Note: the captured RequestContext is not wholly unused today — the success == true branch passes it to MarkDispatched/MarkDispatchedAsync at OutboxProducerMediator.cs:749, 776.)
  • C-7 (Dependency for FR-2). Creating the span Link in FR-2 requires the original publish trace context (the producer/publish span's ActivityContext) to be reachable at confirmation-callback time. Per C-6 this is not available today: the captured construction-time RequestContext is empty of per-publish data, and the producer span has been ended/cleared by the time the asynchronous confirmation fires. How the context is made reachable at callback time is a design decision owned by ADR 0063; the requirement here is only that it be reachable, and that FR-2 degrade gracefully (standalone failure span, Warning log, and breaker trip still occur, but with no Link) when it cannot be.
  • C-8 (Topic source = per-message wire topic; reachability dependency). FR-1 (log) and FR-3 (TripTopic: the interface overload takes RoutingKey — IAmAnOutboxCircuitBreaker.cs:44; the mediator's private overload takes RoutingKey? — OutboxProducerMediator.cs:1168) MUST use the per-message wire topic message.Header.Topic — the exact value the non-confirmation !sent path trips (OutboxProducerMediator.cs:998) — not producer.Publication.Topic. The two diverge for reply/rewritten-topic messages (GetProducerLookupTopic, OutboxProducerMediator.cs:786-807): Header.Topic may be a dynamic reply address while Publication.Topic is the static configured topic, so tripping Publication.Topic would break parity and is rejected as the source. Dependency: the wire topic is NOT carried by the current OnMessagePublished delegate (Action<bool, string>) nor by the captured construction-time RequestContext (C-6), so it must be made reachable at callback time. The producer is positioned to supply it because it already receives the message in its Send/SendAsync call. How the wire topic (and the C-7 trace context) reach the callback is a design decision owned by ADR 0063.

Out of Scope

  • OOS-1. Changing the Sweeper retry semantics or bubbling produce exceptions to callers (explicitly preserved, C-1).
  • OOS-2. Fixing the wrong-typed catch (ProduceException<string, string>) clauses in KafkaMessageProducer.SendWithDelay / SendWithDelayAsync. The producer is IProducer<string, byte[]>, so those catches are effectively dead code. Noted as a known low-risk issue; no behaviour change here.
  • OOS-3. Any new public configuration surface for toggling this observability/breaker behaviour. Behaviour is always on (NFR-5).
  • OOS-4. Changing the success == true (acked) path's delivery semantics is out of scope: the existing "sent" Info log and Outbox mark-dispatched are preserved exactly, and the success branch gains no Warning log and no breaker trip. The success branch does gain the FR-2 confirmation span and the correlation/orphan fix (the previously root-parented MarkDispatched span now nests under it) — that observability change is in scope per FR-2; only the delivery behaviour is frozen.

Acceptance Criteria

Given/When/Then assertions. Each FR maps to at least one AC.

AC-1 (FR-1). Given a confirmation-based producer for topic orders, when it raises OnMessagePublished(false, "abc-123"), then exactly one log entry is emitted at Warning level containing message id abc-123 and topic orders, and no entry at Error or higher is emitted.

AC-2 (FR-2, C-2, C-7). Given the original publish span S1 for abc-123 has already ended and its ActivityContext was made reachable at callback time (per the C-7 mechanism the ADR adopts), when OnMessagePublished(false, "abc-123") fires, then a new standalone Activity is emitted carrying an ActivityLink whose ActivityContext equals S1's context, the original producer span is not mutated, and no failure attributes are written to an unrelated ambient span.

AC-2b (FR-2 degradation). Given no publish span context is reachable at callback time (e.g. tracing disabled, no listener), when OnMessagePublished(false, "abc-123") fires, then a standalone failure Activity is still emitted (carrying message id and topic) with no ActivityLink, and FR-1/FR-3 still occur.

AC-3 (FR-3, C-8, NFR-6). Given the circuit breaker is closed and a message whose wire topic Header.Topic is payments, when OnMessagePublished(false, ...) fires for it, then TripTopic is invoked with the RoutingKey equal to message.Header.Topic (payments) — the identical argument the non-confirmation !sent path passes (OutboxProducerMediator.cs:998).

AC-3b (FR-3 reply-topic parity, C-8). Given a reply message whose Header.Topic was rewritten to a dynamic address reply-xyz distinct from its producer's Publication.Topic, when OnMessagePublished(false, ...) fires for it, then the breaker trips reply-xyz (the wire topic), not Publication.Topic — matching what the non-confirmation path would trip.

AC-4 (FR-4, C-1). Given a deposited message abc-123 that fails to publish, when OnMessagePublished(false, "abc-123") fires, then the Outbox still shows abc-123 as un-dispatched, no exception propagates to the original caller, and the message remains eligible for Sweeper retry.

AC-5 (FR-5). Given a confirmation failure with an empty message id, when OnMessagePublished(false, "") fires, then a single Warning log (id shown as empty/unknown) and an OTel signal are emitted, the breaker is tripped for the associated topic, and no exception is thrown.

AC-6 (FR-6, FR-7). Given the Confluent client throws ProduceException with reason Message size too large and code MsgSizeTooLarge inside KafkaMessagePublisher.PublishMessageAsync, when the publish is attempted, then a Warning log containing both the reason text and the code is emitted, the method returns the synthetic NotPersisted DeliveryResult without rethrowing, and the callback still fires for the failed message.

AC-7 (FR-8). Given a synthetic NotPersisted Kafka report whose headers contain MESSAGE_ID = "abc-123", when KafkaMessageProducer.PublishResults processes it, then OnMessagePublished(false, "abc-123") is fired (not OnMessagePublished(false, string.Empty)).

AC-8 (FR-8 boundary). Given a synthetic NotPersisted Kafka report with no MESSAGE_ID header, when PublishResults processes it, then OnMessagePublished(false, string.Empty) is fired and the downstream shared path behaves per AC-5 (empty-id handling).

AC-9 (FR-9, RMQ). Given a RabbitMQ producer, when the broker nacks message def-456 and OnMessagePublished(false, "def-456") fires, then the shared mediator path emits a Warning log containing def-456, emits an OTel signal, trips the breaker for the RMQ topic, and leaves the message un-dispatched — with no RMQ producer-side code change required.

AC-10 (FR-3 repeated failures / NFR-3). Given multiple messages on the same topic failing via concurrent OnMessagePublished(false, ...) calls (on gateway-dependent threads), when the callbacks run concurrently, then the breaker trip and Warning logs occur without race-induced exceptions or lost log/telemetry entries, and the topic ends up tripped.

AC-11 (NFR-1 negative). Given any success == false confirmation, when logged, then the entry is asserted to be Warning level specifically; a test fails if it is emitted at Error or Critical.

AC-12 (NFR-2 (i)+(ii) non-blocking). Given any confirmation (success or failure), when the callback runs, then no synchronous network I/O is performed on the callback thread for telemetry, and the standalone OTel span is started and stopped within the callback (short-lived).

AC-12b (NFR-2 (iii) no new broker/Outbox await). Given a success == false confirmation, when the failure path runs, then it performs no MarkDispatched/MarkDispatchedAsync or other Outbox call and adds no awaited broker round-trip — the only side effects are the Warning log (FR-1), the standalone span (FR-2), and the breaker trip (FR-3). (Contrast: the success == true path does await MarkDispatched.)

AC-13 (FR-2 success branch + OOS-4 delivery guard). Given a success == true confirmation, when OnMessagePublished(true, "abc-123") fires, then: (a) the existing delivery behaviour is preserved — an informational "sent" log and Outbox mark-dispatched occur; (b) no Warning log and no breaker trip are produced (those stay failure-only); and (c) a standalone confirmation Activity S2 (outcome = confirmed) is emitted carrying an ActivityLink to S1, and the MarkDispatched DB span is parented under S2 (i.e. it is not an orphaned root span). Degraded: when no publish context is reachable, S2 is still emitted without a Link and mark-dispatched still occurs.

AC-14 (NFR-4 error isolation). Given the telemetry/span emission (or other observability handling) throws inside the callback, when OnMessagePublished(false, ...) fires, then the exception does not propagate out of the callback (the producer's invoking thread is not destabilised), the breaker is still tripped (FR-3), and the Warning log is still emitted (FR-1).

Additional Context

  • Source chain (ground truth):

    1. KafkaMessagePublisher.PublishMessageAsync (KafkaMessagePublisher.cs:52-62) catches ProduceException, builds a synthetic DeliveryResult with Status = NotPersisted, invokes the callback, and logs nothing → reason/code lost (gap → FR-6).
    2. Because the exception is swallowed, SendAsync returns normally → sent == true in OutboxProducerMediator.DispatchAsync, so if(!sent) TripTopic(...) (OutboxProducerMediator.cs:998) never runs for confirmation producers → breaker never trips (gap → FR-3).
    3. KafkaMessageProducer.PublishResults (KafkaMessageProducer.cs:364-384) reads MESSAGE_ID only on the Persisted branch; on NotPersisted it fires OnMessagePublished(false, string.Empty), discarding the id even though the synthetic report carries it (gap → FR-8).
    4. OutboxProducerMediator.ConfigureAsyncPublisherCallbackMaybe / ConfigurePublisherCallbackMaybe (OutboxProducerMediator.cs:737-784) handle only success == true; false is silently dropped (gaps → FR-1, FR-2, FR-3).
    5. On success == true, the callback calls MarkDispatched/MarkDispatchedAsync with the empty construction-time RequestContext (C-6) whose .Span is null; the Outbox creates the MarkDispatched DB span with that null parent (RelationDatabaseOutbox.cs CreateDbSpan(..., requestContext?.Span, ...)), and the async callback thread's Activity.Current is also unset — so the span surfaces as a disconnected root span, unlinked from the publish (defect addressed by FR-2's success-branch confirmation span).
    6. Contrast: RmqMessageProducer.cs:480 (async; sync analogous) already invokes OnMessagePublished(false, messageId) with the id on nack → RMQ needs verification only (FR-9).

  • Design decisions folded in: standalone OTel span carrying an ActivityLink to the original publish span on both the success and failure confirmation branches (FR-2, rationale C-2, dependency C-7) — symmetric because the success branch's MarkDispatched Outbox span is currently emitted orphaned (null parent, C-6); the confirmation span fixes that by becoming its parent, so the scope is framed as cleaning up how the confirmation callback reports overall, not a success-path feature add (the Warning log and breaker trip stay failure-only). Circuit breaker trip in scope (FR-3); breaker/log topic = per-message wire topic message.Header.Topic for exact parity with the non-confirmation path including reply/rewritten topics (C-8, NFR-6). Scope = shared mediator fix (FR-1/3/4/5) + Kafka publisher logging (FR-6/7) + Kafka id propagation (FR-8) + RMQ id verification (FR-9) + making the wire topic and original trace context reachable at callback time. The mechanism for that reachability — and any resulting producer-side change beyond verify-only for RMQ — is owned by ADR 0063.

Critical Files for Implementation

  • src/Paramore.Brighter/OutboxProducerMediator.cs — shared callback handling: ConfigureAsyncPublisherCallbackMaybe/ConfigurePublisherCallbackMaybe (~737-784), TripTopic (~998)
  • src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessagePublisher.cs — PublishMessageAsync (~52-62): log swallowed ProduceException
  • src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessageProducer.cs — PublishResults (~364-384): propagate failed message id
  • src/Paramore.Brighter.MessagingGateway.RMQ.Async/RmqMessageProducer.cs and src/Paramore.Brighter.MessagingGateway.RMQ.Sync/RmqMessageProducer.cs — nack path (~480): verification only
  • src/Paramore.Brighter/ISupportPublishConfirmation.cs — the OnMessagePublished contract shared by all confirmation-based producers; the channel by which the wire topic and trace context reach the callback is an ADR decision (C-7/C-8)
  • src/Paramore.Brighter/CircuitBreaker/IAmAnOutboxCircuitBreaker.cs — TripTopic(RoutingKey) (FR-3 target)