mako_engine/dead_letter.rs
1//! Dead-letter sink for unroutable or unprocessable inbound messages.
2//!
3//! BDEW AS4 requires that every received message is either processed or
4//! explicitly refused (CONTRL negative acknowledgement). Messages that
5//! cannot be routed to a workflow — because the PID is unknown, the
6//! conversation is not in-flight, or the format version has no adapter —
7//! must not be silently dropped.
8//!
9//! # Design
10//!
11//! [`DeadLetterSink`] is a synchronous trait that receives a structured
12//! [`DeadLetterReason`] for every rejected message. The synchronous contract
13//! keeps dispatch-path hot code fast; implementations that need async work
14//! (e.g. persisting to a durable DLQ or sending a CONTRL) can use
15//! `tokio::spawn` internally.
16//!
17//! # Implementations
18//!
19//! | Type | Behaviour |
20//! |------|-----------|
21//! | [`LogDeadLetterSink`] | Emits structured `tracing::warn!`; suitable for all deployments |
22//! | [`NoopDeadLetterSink`] | Silently discards; **only for testing** |
23//!
24//! # Wiring
25//!
26//! Pass an implementation to [`EngineBuilder::with_dead_letter_sink`].
27//! The default is [`LogDeadLetterSink`] so unroutable messages are always
28//! visible in the log output without any configuration.
29//!
30//! ```rust
31//! use mako_engine::dead_letter::{AuditContext, DeadLetterReason, DeadLetterSink, LogDeadLetterSink};
32//! use mako_engine::ids::Pid;
33//!
34//! let sink = LogDeadLetterSink;
35//! sink.reject(&DeadLetterReason::UnknownPid { pid: Pid::new(99999), context: AuditContext::now() });
36//! ```
37//!
38//! [`EngineBuilder::with_dead_letter_sink`]: crate::builder::EngineBuilder::with_dead_letter_sink
39
40use std::sync::Arc;
41
42// ── AuditContext ──────────────────────────────────────────────────────────────
43
44/// Structured audit context attached to every dead-letter event.
45///
46/// All fields are `Option` because they are only partially known at rejection
47/// time (e.g. `pid` is not available for a parse failure before the PID is
48/// decoded). Callers fill in as many fields as they have.
49///
50/// These fields map to § 147 AO / GoBD audit-log requirements for AS4 message
51/// rejection events:
52///
53/// | Field | § 147 AO / GoBD requirement |
54/// |---|---|
55/// | `message_type` | Nachrichtentyp (UTILMD, MSCONS, APERAK, …) |
56/// | `release_code` | Releasekennung (S2.1, G1.1, 2.4c, …) |
57/// | `pid` | Prüfidentifikator |
58/// | `sender_eic` | GLN des Absenders |
59/// | `receiver_eic` | GLN des Empfängers |
60/// | `message_ref` | UNH-Referenz |
61/// | `process_id` | Geschäftsvorfallkennung |
62/// | `tenant_id` | Mandant |
63/// | `correlation_id` | AS4 `ConversationId` or similar |
64/// | `timestamp` | Zeitstempel des Eingangs (German local time) |
65#[derive(Debug, Clone)]
66pub struct AuditContext {
67 /// EDIFACT message type (e.g. `"UTILMD"`, `"MSCONS"`, `"APERAK"`).
68 pub message_type: Option<String>,
69 /// BDEW release code (e.g. `"S2.1"`, `"G1.1"`, `"2.4c"`).
70 pub release_code: Option<String>,
71 /// BDEW Prüfidentifikator numeric code.
72 pub pid: Option<crate::ids::Pid>,
73 /// GLN of the AS4 sender.
74 pub sender_eic: Option<String>,
75 /// GLN of the AS4 receiver.
76 pub receiver_eic: Option<String>,
77 /// UNH message reference (interchange + message ref).
78 pub message_ref: Option<String>,
79 /// Internal process / workflow stream ID.
80 pub process_id: Option<String>,
81 /// Tenant identifier (Mandant).
82 pub tenant_id: Option<String>,
83 /// AS4 ConversationId or engine correlation key.
84 pub correlation_id: Option<String>,
85 /// Timestamp of message receipt, in German local time (CET/CEST).
86 pub timestamp: time::OffsetDateTime,
87}
88
89impl AuditContext {
90 /// Create an `AuditContext` with only a timestamp, all other fields `None`.
91 ///
92 /// The timestamp is set to the current wall-clock time in **German local time**
93 /// (CET = UTC+1 in winter, CEST = UTC+2 in summer), satisfying the § 147 AO / GoBD
94 /// requirement for German-timezone audit records.
95 ///
96 /// Use builder-style setters to fill in known fields:
97 /// ```rust
98 /// use mako_engine::dead_letter::AuditContext;
99 /// use mako_engine::ids::Pid;
100 ///
101 /// let ctx = AuditContext::now()
102 /// .with_message_type("UTILMD")
103 /// .with_pid(Pid::new(55001))
104 /// .with_sender_eic("4012345000023");
105 /// ```
106 #[must_use]
107 pub fn now() -> Self {
108 Self {
109 message_type: None,
110 release_code: None,
111 pid: None,
112 sender_eic: None,
113 receiver_eic: None,
114 message_ref: None,
115 process_id: None,
116 tenant_id: None,
117 correlation_id: None,
118 // Use Berlin local time so audit records align with the German
119 // regulatory clock — an off-by-one-hour error at DST transitions
120 // is a reportable BNetzA regulatory violation.
121 timestamp: mako_fristen::berlin_now(),
122 }
123 }
124
125 /// Populate an `AuditContext` from an interchange header and known optional fields.
126 ///
127 /// Fills in `sender_eic`, `receiver_eic`, and `message_ref` (interchange control
128 /// reference) from the parsed UNB header. All remaining fields (pid, process_id,
129 /// tenant_id, correlation_id) are `None` and should be set via builder setters
130 /// when available.
131 ///
132 /// Satisfies the § 147 AO / GoBD requirement that every dead-letter record carries at
133 /// minimum the sender GLN, receiver GLN, and interchange reference.
134 #[must_use]
135 pub fn from_interchange(sender_id: &str, receiver_id: &str, control_ref: &str) -> Self {
136 Self::now()
137 .with_sender_eic(sender_id)
138 .with_receiver_eic(receiver_id)
139 .with_message_ref(control_ref)
140 }
141
142 /// Set the message type.
143 #[must_use]
144 pub fn with_message_type(mut self, mt: impl Into<String>) -> Self {
145 self.message_type = Some(mt.into());
146 self
147 }
148
149 /// Set the BDEW release code.
150 #[must_use]
151 pub fn with_release_code(mut self, rc: impl Into<String>) -> Self {
152 self.release_code = Some(rc.into());
153 self
154 }
155
156 /// Set the Prüfidentifikator.
157 #[must_use]
158 pub fn with_pid(mut self, pid: crate::ids::Pid) -> Self {
159 self.pid = Some(pid);
160 self
161 }
162
163 /// Set the sender GLN.
164 #[must_use]
165 pub fn with_sender_eic(mut self, eic: impl Into<String>) -> Self {
166 self.sender_eic = Some(eic.into());
167 self
168 }
169
170 /// Set the receiver GLN.
171 #[must_use]
172 pub fn with_receiver_eic(mut self, eic: impl Into<String>) -> Self {
173 self.receiver_eic = Some(eic.into());
174 self
175 }
176
177 /// Set the UNH message reference.
178 #[must_use]
179 pub fn with_message_ref(mut self, r: impl Into<String>) -> Self {
180 self.message_ref = Some(r.into());
181 self
182 }
183
184 /// Set the internal process / stream ID.
185 #[must_use]
186 pub fn with_process_id(mut self, id: impl Into<String>) -> Self {
187 self.process_id = Some(id.into());
188 self
189 }
190
191 /// Set the tenant identifier.
192 #[must_use]
193 pub fn with_tenant_id(mut self, id: impl Into<String>) -> Self {
194 self.tenant_id = Some(id.into());
195 self
196 }
197
198 /// Set the AS4 correlation / conversation ID.
199 #[must_use]
200 pub fn with_correlation_id(mut self, id: impl Into<String>) -> Self {
201 self.correlation_id = Some(id.into());
202 self
203 }
204}
205
206impl Default for AuditContext {
207 fn default() -> Self {
208 Self::now()
209 }
210}
211
212// ── DeadLetterReason ──────────────────────────────────────────────────────────
213
214/// Structured reason why an inbound message was rejected.
215///
216/// The variant gives the dispatch path enough information to emit an
217/// actionable CONTRL or log entry. Each variant carries an [`AuditContext`]
218/// with the § 147 AO / GoBD fields required for regulatory audit logging.
219///
220/// Adding new variants is a non-breaking change thanks to `#[non_exhaustive]`.
221#[derive(Debug, Clone)]
222#[non_exhaustive]
223pub enum DeadLetterReason {
224 /// No workflow is registered for this PID in the [`PidRouter`].
225 ///
226 /// The PID is either from a future BDEW release not yet deployed or a
227 /// malformed message. Respond with a CONTRL negative acknowledgement.
228 ///
229 /// [`PidRouter`]: crate::pid_router::PidRouter
230 UnknownPid {
231 /// The numeric Prüfidentifikator that had no registered workflow.
232 pid: crate::ids::Pid,
233 /// § 147 AO / GoBD structured audit context.
234 context: AuditContext,
235 },
236
237 /// No in-flight process matched the inbound `conversation_id`.
238 ///
239 /// This typically means the process completed, was never started, or
240 /// the [`ProcessRegistry`] was lost on restart (see.
241 ///
242 /// [`ProcessRegistry`]: crate::registry::ProcessRegistry
243 UnknownConversation {
244 /// The `conversation_id` from the inbound EDIFACT interchange.
245 conversation_id: String,
246 /// § 147 AO / GoBD structured audit context.
247 context: AuditContext,
248 },
249
250 /// The message's format version has no registered [`MessageAdapter`].
251 ///
252 /// Either the adapter registry is incomplete (see or the sender
253 /// is using a deprecated / future format version.
254 ///
255 /// [`MessageAdapter`]: crate::message_adapter::MessageAdapter
256 VersionMismatch {
257 /// The format version string the adapter registry expected.
258 expected: String,
259 /// The format version string carried in the inbound message.
260 received: String,
261 /// § 147 AO / GoBD structured audit context.
262 context: AuditContext,
263 },
264
265 /// A message with this inbox key was already accepted (AS4 duplicate).
266 ///
267 /// The AS4 sender retries for up to 72 hours. The [`InboxStore`]
268 /// detected the duplicate and the message must not be processed again.
269 ///
270 /// [`InboxStore`]: crate::inbox::InboxStore
271 DuplicateMessage {
272 /// The inbox deduplication key (typically the AS4 `MessageId`).
273 inbox_key: String,
274 /// § 147 AO / GoBD structured audit context.
275 context: AuditContext,
276 },
277
278 /// A workflow or adapter returned a processing error.
279 ///
280 /// The message was routed correctly but could not be processed. Use
281 /// this variant when the failure is definitive (not retriable).
282 ProcessingError {
283 /// Short, human-readable description of the failure.
284 message: String,
285 /// § 147 AO / GoBD structured audit context.
286 context: AuditContext,
287 },
288
289 /// An interchange flagged with UNB DE0035 = 1 (test indicator) was received
290 /// on a production endpoint.
291 ///
292 /// Per Allgemeine Festlegungen V6.1d §3, test interchanges **must not** be
293 /// processed as production. The interchange is rejected at the ingest boundary
294 /// without being forwarded to any workflow.
295 TestMessage {
296 /// § 147 AO / GoBD structured audit context (contains sender, receiver, control_ref).
297 context: AuditContext,
298 },
299
300 /// A business message arrived without the `NAD` party the answer is
301 /// addressed with, or the one the Sparte is resolved from.
302 ///
303 /// BDEW Allgemeine Festlegungen V6.1d §2.13 identifies the *fachliche*
304 /// sender and receiver on message level in `NAD+MS` / `NAD+MR` DE 3035 and
305 /// states the approach applies "für alle EDI@Energy EDIFACT Nachrichten und
306 /// -dateien einheitlich". Both are load-bearing here: the sender is who the
307 /// answer and the APERAK go back to, and the receiver is which of the
308 /// operator's own MP-IDs — and therefore which Sparte and which Marktrolle —
309 /// the message is addressed to.
310 ///
311 /// Substituting an empty MP-ID produces a process whose answer is addressed
312 /// to nobody and whose Sparte silently defaults, so the message is refused
313 /// at the boundary instead. CONTRL carries no `NAD` at all — it is a
314 /// UN/EDIFACT syntax acknowledgement rather than an EDI@Energy business
315 /// message — and is exempt.
316 MissingInterchangeParty {
317 /// `"MS"` or `"MR"` — which of the two was absent or empty.
318 qualifier: &'static str,
319 /// § 147 AO / GoBD structured audit context.
320 context: AuditContext,
321 },
322
323 /// The PID resolved to a workflow, but the ingest dispatcher has no arm
324 /// for it — so `makod` claimed to route the message and then dropped it.
325 ///
326 /// Always a coverage bug on this side, never a defect in the sender's
327 /// message: [`PidRouter`] answered, the transport acknowledged, and the
328 /// business payload went nowhere. It is recorded rather than logged because
329 /// an acknowledged inbound message that produced no process is exactly what
330 /// § 147 AO / GoBD require a trace of.
331 ///
332 /// [`PidRouter`]: crate::pid_router::PidRouter
333 NotDispatchable {
334 /// The workflow name [`PidRouter`] resolved the PID to.
335 ///
336 /// [`PidRouter`]: crate::pid_router::PidRouter
337 workflow_name: String,
338 /// The Prüfidentifikator carried by the message.
339 pid: crate::ids::Pid,
340 /// Machine-readable skip reason from the dispatcher.
341 reason: String,
342 /// § 147 AO / GoBD structured audit context.
343 context: AuditContext,
344 },
345
346 /// The outbox delivery worker gave up after exhausting all retry attempts.
347 ///
348 /// The message was re-queued `max_attempts` times and never successfully
349 /// delivered to the AS4 endpoint (or ERP webhook). The message is removed
350 /// from the outbox and recorded here for regulatory audit.
351 OutboxExhausted {
352 /// The outbox message ID of the undeliverable message.
353 message_id: crate::ids::OutboxMessageId,
354 /// The message type (e.g. `"APERAK"`, `"CONTRL"`).
355 message_type: String,
356 /// The intended recipient GLN.
357 recipient: String,
358 /// The last error returned by the AS4 sender.
359 last_error: String,
360 /// How many delivery attempts were made.
361 attempts: u32,
362 },
363}
364
365impl DeadLetterReason {
366 /// Short label identifying the rejection category.
367 ///
368 /// Suitable for structured log fields and metric labels.
369 #[must_use]
370 pub fn label(&self) -> &'static str {
371 match self {
372 Self::UnknownPid { .. } => "unknown_pid",
373 Self::UnknownConversation { .. } => "unknown_conversation",
374 Self::VersionMismatch { .. } => "version_mismatch",
375 Self::DuplicateMessage { .. } => "duplicate_message",
376 Self::ProcessingError { .. } => "processing_error",
377 Self::TestMessage { .. } => "test_message",
378 Self::MissingInterchangeParty { .. } => "missing_interchange_party",
379 Self::NotDispatchable { .. } => "not_dispatchable",
380 Self::OutboxExhausted { .. } => "outbox_exhausted",
381 }
382 }
383
384 /// Return the [`AuditContext`] embedded in this reason, if present.
385 ///
386 /// `OutboxExhausted` does not carry an `AuditContext` because it refers
387 /// to an outbound message (not an inbound AS4 message).
388 #[must_use]
389 pub fn audit_context(&self) -> Option<&AuditContext> {
390 match self {
391 Self::UnknownPid { context, .. }
392 | Self::UnknownConversation { context, .. }
393 | Self::VersionMismatch { context, .. }
394 | Self::DuplicateMessage { context, .. }
395 | Self::ProcessingError { context, .. }
396 | Self::TestMessage { context, .. }
397 | Self::MissingInterchangeParty { context, .. }
398 | Self::NotDispatchable { context, .. } => Some(context),
399 Self::OutboxExhausted { .. } => None,
400 }
401 }
402}
403
404impl std::fmt::Display for DeadLetterReason {
405 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
406 match self {
407 Self::UnknownPid { pid, .. } => write!(f, "unknown PID {pid}"),
408 Self::UnknownConversation {
409 conversation_id, ..
410 } => {
411 write!(f, "unknown conversation {conversation_id}")
412 }
413 Self::VersionMismatch {
414 expected, received, ..
415 } => write!(
416 f,
417 "version mismatch: expected {expected}, received {received}"
418 ),
419 Self::DuplicateMessage { inbox_key, .. } => write!(f, "duplicate message {inbox_key}"),
420 Self::ProcessingError { message, .. } => write!(f, "processing error: {message}"),
421 Self::TestMessage { context } => write!(
422 f,
423 "test interchange rejected (DE0035=1): sender={}, receiver={}, ref={}",
424 context.sender_eic.as_deref().unwrap_or(""),
425 context.receiver_eic.as_deref().unwrap_or(""),
426 context.message_ref.as_deref().unwrap_or(""),
427 ),
428 Self::MissingInterchangeParty { qualifier, .. } => write!(
429 f,
430 "message carries no NAD+{qualifier} party (Allgemeine Festlegungen §2.13)"
431 ),
432 Self::NotDispatchable {
433 workflow_name,
434 pid,
435 reason,
436 ..
437 } => write!(
438 f,
439 "PID {pid} routed to workflow {workflow_name} but no dispatch arm handled it ({reason})"
440 ),
441 Self::OutboxExhausted {
442 message_id,
443 message_type,
444 recipient,
445 attempts,
446 ..
447 } => write!(
448 f,
449 "outbox exhausted after {attempts} attempts: {message_type} → {recipient} (id={message_id})"
450 ),
451 }
452 }
453}
454
455// ── DeadLetterSink trait ──────────────────────────────────────────────────────
456
457/// Receives messages that cannot be routed or processed.
458///
459/// Implement this trait to:
460/// - Emit CONTRL negative acknowledgements for unroutable messages
461/// - Persist rejections to a durable dead-letter queue for manual review
462/// - Trigger alerts when duplicate-message counts exceed a threshold
463///
464/// The method is **synchronous**. Implementations that require async work
465/// (network calls, database writes) must use `tokio::spawn` internally.
466///
467/// # Default
468///
469/// The default dead-letter sink is [`LogDeadLetterSink`], which emits
470/// `tracing::warn!` events. Override with
471/// [`EngineBuilder::with_dead_letter_sink`] to add CONTRL dispatch or
472/// persistent DLQ storage.
473///
474/// [`EngineBuilder::with_dead_letter_sink`]: crate::builder::EngineBuilder::with_dead_letter_sink
475pub trait DeadLetterSink: Send + Sync + 'static {
476 /// Record a rejected message.
477 ///
478 /// Called by the dispatch path synchronously, before the inbound
479 /// message is acknowledged at the AS4 transport layer. Must not block.
480 fn reject(&self, reason: &DeadLetterReason);
481}
482
483// ── LogDeadLetterSink ─────────────────────────────────────────────────────────
484
485/// A [`DeadLetterSink`] that emits a structured `tracing::warn!` event for
486/// every rejected message.
487///
488/// Suitable for all deployment tiers. In production, combine with a
489/// `tracing` subscriber that forwards `warn`-level events to your alert
490/// pipeline (Loki, CloudWatch, etc.).
491///
492/// This is the **default** dead-letter sink in [`EngineBuilder`].
493///
494/// [`EngineBuilder`]: crate::builder::EngineBuilder
495#[derive(Debug, Clone, Default)]
496pub struct LogDeadLetterSink;
497
498impl DeadLetterSink for LogDeadLetterSink {
499 fn reject(&self, reason: &DeadLetterReason) {
500 // Increment Prometheus counter for every rejection, regardless of
501 // which sink is wired — mirrors SlateDbDeadLetterSink behaviour so
502 // alerting works in non-SlateDB and smoke environments too.
503 crate::metrics::EngineMetrics::global().dead_letter_recorded(reason.label());
504 // Emit all § 147 AO / GoBD structured audit fields when available.
505 if let Some(ctx) = reason.audit_context() {
506 tracing::warn!(
507 reason = reason.label(),
508 message_type = ctx.message_type.as_deref().unwrap_or(""),
509 release_code = ctx.release_code.as_deref().unwrap_or(""),
510 pid = ctx.pid.map_or(0, crate::ids::Pid::as_u32),
511 sender_eic = ctx.sender_eic.as_deref().unwrap_or(""),
512 receiver_eic = ctx.receiver_eic.as_deref().unwrap_or(""),
513 message_ref = ctx.message_ref.as_deref().unwrap_or(""),
514 process_id = ctx.process_id.as_deref().unwrap_or(""),
515 tenant_id = ctx.tenant_id.as_deref().unwrap_or(""),
516 correlation_id = ctx.correlation_id.as_deref().unwrap_or(""),
517 %ctx.timestamp,
518 "dead letter: {reason}",
519 );
520 } else {
521 // OutboxExhausted has no inbound audit context; log its own fields.
522 match reason {
523 DeadLetterReason::OutboxExhausted {
524 message_id,
525 message_type,
526 recipient,
527 last_error,
528 attempts,
529 } => {
530 tracing::error!(
531 %message_id,
532 message_type,
533 recipient,
534 last_error,
535 attempts,
536 reason = reason.label(),
537 "dead letter: outbox exhausted — message removed after max delivery \
538 attempts; manual intervention required to deliver this message",
539 );
540 }
541 _ => {
542 tracing::warn!(reason = reason.label(), "dead letter: {reason}");
543 }
544 }
545 }
546 }
547}
548
549// ── NoopDeadLetterSink ────────────────────────────────────────────────────────
550
551/// A [`DeadLetterSink`] that silently discards all rejected messages.
552///
553/// **Use only in unit tests** where dead-letter events are not the subject
554/// under test. Using this in production means unroutable messages are lost
555/// without any diagnostic output, violating BDEW AS4 requirements.
556#[derive(Debug, Clone, Default)]
557#[must_use = "NoopDeadLetterSink discards all rejections; use LogDeadLetterSink in production"]
558#[cfg_attr(
559 not(any(test, feature = "testing")),
560 deprecated = "NoopDeadLetterSink must not be used in production builds; use LogDeadLetterSink instead"
561)]
562pub struct NoopDeadLetterSink;
563
564#[cfg(any(test, feature = "testing"))]
565impl DeadLetterSink for NoopDeadLetterSink {
566 fn reject(&self, _reason: &DeadLetterReason) {}
567}
568
569// ── ArcDeadLetterSink ─────────────────────────────────────────────────────────
570
571/// Blanket implementation so `Arc<T>` is a `DeadLetterSink` whenever `T` is.
572///
573/// This allows passing `Arc<LogDeadLetterSink>` or `Arc<dyn DeadLetterSink>`
574/// wherever a `DeadLetterSink` is expected without an extra wrapper.
575impl<T: DeadLetterSink> DeadLetterSink for Arc<T> {
576 fn reject(&self, reason: &DeadLetterReason) {
577 self.as_ref().reject(reason);
578 }
579}
580
581#[cfg(test)]
582mod tests {
583 use super::*;
584
585 #[test]
586 fn dead_letter_reason_labels() {
587 assert_eq!(
588 DeadLetterReason::UnknownPid {
589 pid: crate::ids::Pid::new(55001),
590 context: AuditContext::now()
591 }
592 .label(),
593 "unknown_pid"
594 );
595 assert_eq!(
596 DeadLetterReason::UnknownConversation {
597 conversation_id: "abc".into(),
598 context: AuditContext::now(),
599 }
600 .label(),
601 "unknown_conversation"
602 );
603 assert_eq!(
604 DeadLetterReason::VersionMismatch {
605 expected: "FV2025-10-01".into(),
606 received: "FV2026-10-01".into(),
607 context: AuditContext::now(),
608 }
609 .label(),
610 "version_mismatch"
611 );
612 assert_eq!(
613 DeadLetterReason::DuplicateMessage {
614 inbox_key: "msg-1".into(),
615 context: AuditContext::now(),
616 }
617 .label(),
618 "duplicate_message"
619 );
620 assert_eq!(
621 DeadLetterReason::ProcessingError {
622 message: "invalid state".into(),
623 context: AuditContext::now(),
624 }
625 .label(),
626 "processing_error"
627 );
628 }
629
630 #[test]
631 fn log_sink_does_not_panic() {
632 let sink = LogDeadLetterSink;
633 sink.reject(&DeadLetterReason::UnknownPid {
634 pid: crate::ids::Pid::new(99999),
635 context: AuditContext::now(),
636 });
637 sink.reject(&DeadLetterReason::UnknownConversation {
638 conversation_id: "conv-123".into(),
639 context: AuditContext::now(),
640 });
641 sink.reject(&DeadLetterReason::VersionMismatch {
642 expected: "FV2025-10-01".into(),
643 received: "FV2026-10-01".into(),
644 context: AuditContext::now(),
645 });
646 sink.reject(&DeadLetterReason::DuplicateMessage {
647 inbox_key: "msg-42".into(),
648 context: AuditContext::now(),
649 });
650 sink.reject(&DeadLetterReason::ProcessingError {
651 message: "workflow rejected command".into(),
652 context: AuditContext::now(),
653 });
654 }
655
656 #[test]
657 fn noop_sink_does_not_panic() {
658 let sink = NoopDeadLetterSink;
659 sink.reject(&DeadLetterReason::UnknownPid {
660 pid: crate::ids::Pid::new(55001),
661 context: AuditContext::now(),
662 });
663 }
664
665 #[test]
666 fn arc_blanket_impl_works() {
667 let sink: Arc<LogDeadLetterSink> = Arc::new(LogDeadLetterSink);
668 sink.reject(&DeadLetterReason::UnknownPid {
669 pid: crate::ids::Pid::new(1),
670 context: AuditContext::now(),
671 });
672 }
673
674 #[test]
675 fn dead_letter_reason_display() {
676 assert_eq!(
677 DeadLetterReason::UnknownPid {
678 pid: crate::ids::Pid::new(55001),
679 context: AuditContext::now()
680 }
681 .to_string(),
682 "unknown PID 55001"
683 );
684 assert!(
685 DeadLetterReason::VersionMismatch {
686 expected: "FV2025-10-01".into(),
687 received: "FV2026-10-01".into(),
688 context: AuditContext::now(),
689 }
690 .to_string()
691 .contains("version mismatch")
692 );
693 }
694
695 #[test]
696 fn audit_context_builder() {
697 let ctx = AuditContext::now()
698 .with_message_type("UTILMD")
699 .with_pid(crate::ids::Pid::new(55001))
700 .with_sender_eic("4012345000023")
701 .with_receiver_eic("9900357000004")
702 .with_message_ref("00001")
703 .with_tenant_id("tenant-a")
704 .with_correlation_id("conv-xyz");
705
706 assert_eq!(ctx.message_type.as_deref(), Some("UTILMD"));
707 assert_eq!(ctx.pid, Some(crate::ids::Pid::new(55001)));
708 assert_eq!(ctx.sender_eic.as_deref(), Some("4012345000023"));
709 assert_eq!(ctx.correlation_id.as_deref(), Some("conv-xyz"));
710 }
711
712 #[test]
713 fn audit_context_returned_for_inbound_reasons() {
714 let r = DeadLetterReason::UnknownPid {
715 pid: crate::ids::Pid::new(99),
716 context: AuditContext::now().with_pid(crate::ids::Pid::new(99)),
717 };
718 assert!(r.audit_context().is_some());
719 }
720}