From 712f9df01725dfaae4b978d0eefaffee7efe9c3d Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Tue, 7 Jul 2026 02:36:47 -0500 Subject: [PATCH] feat: add structured failure logs Implements [[tasks/distributed-structured-failure-logs-1]] --- Cargo.toml | 3 +- docs/metrics.md | 5 + docs/observability.md | 86 +- src/bus/runner.rs | 249 +++++- src/failure_log.rs | 1128 ++++++++++++++++++++++++++ src/lib.rs | 1 + src/microsvc/grpc.rs | 130 ++- src/microsvc/http.rs | 110 ++- src/microsvc/knative_ingress.rs | 109 ++- src/microsvc/service.rs | 66 +- src/outbox_worker/outbox_dispatch.rs | 306 ++++++- src/outbox_worker/publish_hook.rs | 1 + src/telemetry.rs | 5 + 13 files changed, 2145 insertions(+), 54 deletions(-) create mode 100644 src/failure_log.rs diff --git a/Cargo.toml b/Cargo.toml index 7c36648..d2eadeb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,6 +30,7 @@ path = "src/lib.rs" default = [] emitter = ["dep:event-emitter-rs"] metrics = [] +failure-logs = ["dep:tracing"] http = ["dep:axum", "dep:reqwest", "dep:tokio"] grpc = ["dep:tonic", "dep:tonic-prost", "dep:prost", "dep:tokio"] postgres = ["dep:sqlx", "dep:tokio", "sqlx/postgres", "sqlx/migrate", "sqlx/runtime-tokio"] @@ -37,7 +38,7 @@ sqlite = ["dep:sqlx", "dep:tokio", "sqlx/migrate", "sqlx/runtime-tokio", "sqlx/s nats = ["dep:async-nats", "dep:futures", "dep:tokio"] rabbitmq = ["dep:lapin", "dep:futures", "dep:tokio"] kafka = ["dep:rdkafka", "dep:tokio"] -otel = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:tracing", "dep:tracing-opentelemetry"] +otel = ["failure-logs", "dep:opentelemetry", "dep:opentelemetry_sdk", "dep:tracing-opentelemetry"] [dependencies] async-nats = { version = "0.49", optional = true } diff --git a/docs/metrics.md b/docs/metrics.md index 0c08aac..8c1a750 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -93,6 +93,11 @@ labels, and numeric samples only; diagnostics must not add payloads, metadata, trace ids, aggregate ids, user ids, raw HTTP targets, or request ids to that shape. +Structured failure diagnostics are emitted as `tracing` events by the optional +`failure-logs` feature, not as metrics. Do not add trace ids, message ids, +users, aggregate identifiers, raw paths, payloads, error strings, or SQL state +labels to Prometheus series to mirror those logs. + ## Boundaries The `metrics` feature owns Prometheus text exposition for framework metrics. It diff --git a/docs/observability.md b/docs/observability.md index 8fc2db1..55f429f 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -92,6 +92,80 @@ Future spans should use the same helper path so new attributes are reviewed in one place. Do not add payload fields, user ids, aggregate ids, or raw metadata values as framework span attributes. +## Structured Failure Events + +The `failure-logs` feature emits framework-owned `tracing` events for failures +and deliberate drops. It does not install a subscriber, formatter, exporter, or +global logging backend; service binaries keep control of routing and retention. +The `otel` feature enables `failure-logs` as well, so span-enabled services also +emit the same structured failure vocabulary. + +```toml +[dependencies] +distributed = { version = "0.1", features = ["http", "failure-logs"] } +``` + +All failure events use target `distributed.failure`, field +`event.name = "distributed.failure"`, and +`distributed.failure.schema_version = 1`. + +Core fields: + +- `distributed.service.name` +- `distributed.component`: `microsvc`, `http`, `grpc`, `knative`, + `transport`, `outbox`, or `repository` +- `distributed.operation`: `dispatch`, `handler`, `ingress`, `receive`, + `settle`, `publish`, `claim`, or `commit` +- `distributed.message.kind`: `command` or `event` +- `distributed.message.name`: registered bounded name or `unknown` +- `messaging.message.id_hash`: stable hash only, never the raw id +- `distributed.correlation_id` and `distributed.causation_id` +- `trace.trace_id`, `trace.parent_span_id`, `trace.trace_flags`, + `trace.traceparent_valid`, and `trace.tracestate_present` +- `error.type`, `error.category`, `error.retry_class`, and sanitized + `error.message` +- `distributed.failure.action`: `nack`, `dead_letter`, `park`, + `log_and_ack`, `stop`, `release`, `fail`, `return_400`, `return_401`, + `return_404`, `return_422`, `return_500`, `return_503`, `recv_error`, + `settle_ack`, `settle_nack`, `settle_dead_letter`, or `settle_park` + +Boundary-specific fields are populated when relevant: + +- `http.response.status_code` +- `rpc.grpc.status_code` +- `messaging.system` +- `outbox.attempt`, `outbox.max_attempts`, `outbox.status`, + `outbox.source_aggregate_type`, and `outbox.source_sequence` +- `payload.size_bytes`, `payload.content_type`, and `payload.codec` + +Example event shape: + +```text +target=distributed.failure event.name=distributed.failure +distributed.failure.schema_version=1 +distributed.service.name=orders +distributed.component=http +distributed.operation=ingress +distributed.message.kind=command +distributed.message.name=orders.create +error.type=RepositoryError::StorageRetryable +error.category=storage +error.retry_class=retryable +error.message="storage error (retryable) during load stream: connection refused" +distributed.failure.action=return_500 +http.response.status_code=500 +trace.trace_id=4bf92f3577b34da6a3ce929d0e0e4736 +trace.tracestate_present=true +payload.size_bytes=42 +payload.content_type=application/json +``` + +Payloads, decoded handler inputs, request bodies, arbitrary headers, arbitrary +metadata values, session maps, cookies, tokens, secrets, DB URLs, raw SQL, raw +message ids, raw outbox ids, source aggregate ids, and raw `tracestate` are not +emitted. Message ids are hashed by default; `tracestate` is reduced to a boolean +presence field. + Recommended environment variables for OTLP exporters: ```text @@ -170,10 +244,10 @@ The `metrics` feature records framework metrics and renders Prometheus text. It does not expose OpenTelemetry metrics or request-level HTTP telemetry. The `otel` feature creates framework spans and parents them from W3C metadata -when available. It does not install subscribers, exporters, sampling rules, or -resource attributes. +when available. It also enables `failure-logs`. It does not install subscribers, +exporters, sampling rules, or resource attributes. -Future `logs` or private `diagnostics` features should build on the same -bounded telemetry vocabulary. Logs may carry structured failure context, but -must not log payloads or secrets by default. Diagnostics should read typed -snapshots of framework state rather than parsing public Prometheus text. +Future private `diagnostics` features should build on the same bounded +telemetry vocabulary and may consume the sanitized failure classification. +Diagnostics should read typed snapshots of framework state rather than parsing +public Prometheus text. diff --git a/src/bus/runner.rs b/src/bus/runner.rs index c95c62b..33ae40b 100644 --- a/src/bus/runner.rs +++ b/src/bus/runner.rs @@ -13,6 +13,10 @@ use std::sync::Arc; use super::source::{MessageSource, ReceivedMessage}; use super::{FailureAction, MessageRouter, RunOptions, TransportError, TransportErrorKind}; use super::{Message, MessageKind}; +use crate::failure_log::{ + FailureAction as FailureLogAction, FailureCategory, FailureComponent, FailureMessageFields, + FailureOperation, FailureRecord, FailureRetryClass, +}; /// Run the receive loop for a direct transport source. /// @@ -71,6 +75,8 @@ where let action = options.failure_policy.resolve(error); record_transport_failure(service, transport, error.kind(), action); let kind = received.message().kind; + let message = FailureMessageFields::from_message(received.message()); + emit_decode_failure(service, transport, message.clone(), error, action); match action { FailureAction::Nack => { let reason = error.to_string(); @@ -78,6 +84,7 @@ where service, transport, kind, + message, crate::telemetry::transport_outcome::NACK, crate::telemetry::transport_outcome::NACK, || received.nack(&reason), @@ -90,6 +97,7 @@ where service, transport, kind, + message, crate::telemetry::transport_outcome::DEAD_LETTER, crate::telemetry::transport_outcome::DEAD_LETTER, || received.dead_letter(&reason), @@ -102,6 +110,7 @@ where service, transport, kind, + message, crate::telemetry::transport_outcome::PARK, crate::telemetry::transport_outcome::PARK, || received.park(&reason), @@ -109,11 +118,11 @@ where .await?; } FailureAction::LogAndAck => { - eprintln!("[bus::runner] dropping undecodable message after permanent failure: {error}"); settle_and_record( service, transport, kind, + message, crate::telemetry::transport_outcome::ACK, crate::telemetry::transport_outcome::LOG_AND_ACK, || received.ack(), @@ -132,6 +141,7 @@ where service, transport, kind, + FailureMessageFields::from_message(received.message()), crate::telemetry::transport_outcome::ACK, crate::telemetry::transport_outcome::IGNORED, || received.ack(), @@ -142,10 +152,12 @@ where let kind = received.message().kind; match dispatch(router.as_ref(), &options, received.message()).await { Ok(()) => { + let message = FailureMessageFields::from_message(received.message()); settle_and_record( service, transport, kind, + message, crate::telemetry::transport_outcome::ACK, crate::telemetry::transport_outcome::ACK, || received.ack(), @@ -155,11 +167,14 @@ where Err(error) => match options.failure_policy.resolve(&error) { action @ FailureAction::Nack => { record_transport_failure(service, transport, error.kind(), action); + let message = FailureMessageFields::from_message(received.message()); + emit_transport_failure(service, transport, message.clone(), &error, action); let reason = error.to_string(); settle_and_record( service, transport, kind, + message, crate::telemetry::transport_outcome::NACK, crate::telemetry::transport_outcome::NACK, || received.nack(&reason), @@ -168,11 +183,14 @@ where } action @ FailureAction::DeadLetter => { record_transport_failure(service, transport, error.kind(), action); + let message = FailureMessageFields::from_message(received.message()); + emit_transport_failure(service, transport, message.clone(), &error, action); let reason = error.to_string(); settle_and_record( service, transport, kind, + message, crate::telemetry::transport_outcome::DEAD_LETTER, crate::telemetry::transport_outcome::DEAD_LETTER, || received.dead_letter(&reason), @@ -181,11 +199,14 @@ where } action @ FailureAction::Park => { record_transport_failure(service, transport, error.kind(), action); + let message = FailureMessageFields::from_message(received.message()); + emit_transport_failure(service, transport, message.clone(), &error, action); let reason = error.to_string(); settle_and_record( service, transport, kind, + message, crate::telemetry::transport_outcome::PARK, crate::telemetry::transport_outcome::PARK, || received.park(&reason), @@ -199,14 +220,19 @@ where error.kind(), FailureAction::LogAndAck, ); - eprintln!( - "[bus::runner] dropping message '{}' after permanent failure: {error}", - received.message().name() + let message = FailureMessageFields::from_message(received.message()); + emit_transport_failure( + service, + transport, + message.clone(), + &error, + FailureAction::LogAndAck, ); settle_and_record( service, transport, kind, + message, crate::telemetry::transport_outcome::ACK, crate::telemetry::transport_outcome::LOG_AND_ACK, || received.ack(), @@ -215,6 +241,13 @@ where } FailureAction::Stop => { record_transport_failure(service, transport, error.kind(), FailureAction::Stop); + emit_transport_failure( + service, + transport, + FailureMessageFields::from_message(received.message()), + &error, + FailureAction::Stop, + ); return Err(error); } }, @@ -227,6 +260,7 @@ async fn settle_and_record( service: Option<&str>, transport: &str, kind: MessageKind, + message: FailureMessageFields, settle_action: &'static str, outcome: &'static str, settle: F, @@ -247,6 +281,7 @@ where error.kind(), crate::telemetry::settle_failure_action(settle_action), ); + emit_transport_settle_failure(service, transport, message, settle_action, &error); Err(error) } } @@ -266,6 +301,15 @@ async fn recv_next( error.kind(), crate::telemetry::failure_action::RECV_ERROR, ); + FailureRecord::from_transport_error( + FailureComponent::Transport, + FailureOperation::Receive, + FailureLogAction::RecvError, + &error, + ) + .with_service(service) + .with_transport(transport) + .emit(); Err(error) } } @@ -364,6 +408,86 @@ impl IntoFailureActionLabel for &'static str { } } +fn emit_decode_failure( + service: Option<&str>, + transport: &str, + message: FailureMessageFields, + error: &TransportError, + action: FailureAction, +) { + FailureRecord::new( + FailureComponent::Transport, + FailureOperation::Receive, + FailureCategory::Decode, + failure_log_action(action), + FailureRetryClass::from(error.kind()), + "TransportError::Decode", + error.message(), + ) + .with_service(service) + .with_transport(transport) + .with_message(message) + .emit(); +} + +fn emit_transport_failure( + service: Option<&str>, + transport: &str, + message: FailureMessageFields, + error: &TransportError, + action: FailureAction, +) { + FailureRecord::from_transport_error( + FailureComponent::Transport, + FailureOperation::Receive, + failure_log_action(action), + error, + ) + .with_service(service) + .with_transport(transport) + .with_message(message) + .emit(); +} + +fn emit_transport_settle_failure( + service: Option<&str>, + transport: &str, + message: FailureMessageFields, + settle_action: &'static str, + error: &TransportError, +) { + FailureRecord::from_transport_error( + FailureComponent::Transport, + FailureOperation::Settle, + settle_failure_log_action(settle_action), + error, + ) + .with_service(service) + .with_transport(transport) + .with_message(message) + .emit(); +} + +fn failure_log_action(action: FailureAction) -> FailureLogAction { + match action { + FailureAction::Nack => FailureLogAction::Nack, + FailureAction::DeadLetter => FailureLogAction::DeadLetter, + FailureAction::Park => FailureLogAction::Park, + FailureAction::LogAndAck => FailureLogAction::LogAndAck, + FailureAction::Stop => FailureLogAction::Stop, + } +} + +fn settle_failure_log_action(settle_action: &'static str) -> FailureLogAction { + match settle_action { + crate::telemetry::transport_outcome::ACK => FailureLogAction::SettleAck, + crate::telemetry::transport_outcome::NACK => FailureLogAction::SettleNack, + crate::telemetry::transport_outcome::DEAD_LETTER => FailureLogAction::SettleDeadLetter, + crate::telemetry::transport_outcome::PARK => FailureLogAction::SettlePark, + _ => FailureLogAction::SettleAck, + } +} + #[cfg(test)] mod tests { use super::*; @@ -911,6 +1035,123 @@ mod tests { ); } + #[cfg(feature = "failure-logs")] + #[test] + fn failure_logs_cover_runner_failure_paths_with_consistent_fields() { + fn assert_event( + events: &[crate::failure_log::testing::CapturedEvent], + action: &str, + retry_class: &str, + operation: &str, + ) { + let event = events + .iter() + .find(|event| { + event.field("distributed.failure.action") == Some(action) + && event.field("error.retry_class") == Some(retry_class) + && event.field("distributed.operation") == Some(operation) + }) + .unwrap_or_else(|| { + panic!( + "missing action {action} with retry class {retry_class} and operation {operation}: {events:?}" + ) + }); + assert_eq!(event.field("distributed.failure.schema_version"), Some("1")); + assert_eq!(event.field("distributed.component"), Some("transport")); + assert_eq!(event.field("distributed.operation"), Some(operation)); + assert_eq!(event.field("distributed.message.kind"), Some("event")); + assert_eq!(event.field("messaging.system"), Some("unknown")); + assert_eq!(event.field("error.retry_class"), Some(retry_class)); + } + + { + let capture = crate::failure_log::testing::capture_failures(); + let result = run( + vec![event_message("retryable", Some("retryable-message-id"))], + RunOptions::idempotent(), + ); + assert!(result.outcome.is_ok()); + assert_event(&capture.failure_events(), "nack", "retryable", "receive"); + let rendered = format!("{:?}", capture.failure_events()); + assert!(!rendered.contains("retryable-message-id")); + } + + { + let capture = crate::failure_log::testing::capture_failures(); + let result = run( + vec![event_message("permanent", None)], + RunOptions::idempotent(), + ); + assert!(result.outcome.is_ok()); + assert_event( + &capture.failure_events(), + "dead_letter", + "permanent", + "receive", + ); + } + + { + let capture = crate::failure_log::testing::capture_failures(); + let result = run_decode_error(vec![event_message("", None)], RunOptions::idempotent()); + assert!(result.outcome.is_ok()); + assert_event( + &capture.failure_events(), + "dead_letter", + "permanent", + "receive", + ); + let decode = capture + .failure_events() + .into_iter() + .find(|event| event.field("error.category") == Some("decode")) + .expect("decode failure event"); + assert_eq!(decode.field("error.type"), Some("TransportError::Decode")); + } + + { + let capture = crate::failure_log::testing::capture_failures(); + let result = run_with( + vec![event_message("ok", None)], + RunOptions::idempotent(), + false, + false, + ); + assert!(result.outcome.expect_err("settle failure").is_retryable()); + assert_event( + &capture.failure_events(), + "settle_ack", + "retryable", + "settle", + ); + } + + { + let capture = crate::failure_log::testing::capture_failures(); + let result = run( + vec![event_message("permanent", None)], + RunOptions::idempotent().with_failure_policy(FailurePolicy::LogAndAck), + ); + assert!(result.outcome.is_ok()); + assert_event( + &capture.failure_events(), + "log_and_ack", + "permanent", + "receive", + ); + } + + { + let capture = crate::failure_log::testing::capture_failures(); + let result = run( + vec![event_message("permanent", None)], + RunOptions::idempotent().with_failure_policy(FailurePolicy::Stop), + ); + assert!(result.outcome.expect_err("stop failure").is_permanent()); + assert_event(&capture.failure_events(), "stop", "permanent", "receive"); + } + } + #[test] fn run_source_future_is_send() { // Guards the documented multi-threaded-executor contract for the common diff --git a/src/failure_log.rs b/src/failure_log.rs new file mode 100644 index 0000000..c053ce8 --- /dev/null +++ b/src/failure_log.rs @@ -0,0 +1,1128 @@ +//! Structured failure event schema and sanitization. + +use std::error::Error; + +use crate::bus::{Message, MessageKind, TransportError, TransportErrorKind}; +use crate::lock::RetryClass; +use crate::microsvc::HandlerError; +use crate::repository::RepositoryError; +use crate::trace_context::{is_valid_traceparent, TraceContext}; + +pub(crate) const SCHEMA_VERSION: u64 = 1; + +const UNKNOWN: &str = "unknown"; +const ID_LIMIT: usize = 128; +const MESSAGE_LIMIT: usize = 128; +const ERROR_LIMIT: usize = 512; +const TYPE_LIMIT: usize = 128; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[expect( + dead_code, + reason = "schema vocabulary includes repository even when current emitters report through their owning boundary" +)] +pub(crate) enum FailureComponent { + Microsvc, + Http, + Grpc, + Knative, + Transport, + Outbox, + Repository, +} + +impl FailureComponent { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Microsvc => "microsvc", + Self::Http => "http", + Self::Grpc => "grpc", + Self::Knative => "knative", + Self::Transport => "transport", + Self::Outbox => "outbox", + Self::Repository => "repository", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[expect( + dead_code, + reason = "schema vocabulary reserves handler and commit operations for framework failure boundaries" +)] +pub(crate) enum FailureOperation { + Dispatch, + Handler, + Ingress, + Receive, + Settle, + Publish, + Claim, + Commit, +} + +impl FailureOperation { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Dispatch => "dispatch", + Self::Handler => "handler", + Self::Ingress => "ingress", + Self::Receive => "receive", + Self::Settle => "settle", + Self::Publish => "publish", + Self::Claim => "claim", + Self::Commit => "commit", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[expect( + dead_code, + reason = "schema vocabulary reserves unknown for future unmapped framework failures" +)] +pub(crate) enum FailureCategory { + Routing, + Decode, + Validation, + Auth, + Guard, + Repository, + Storage, + Transport, + OutboxPublish, + OutboxSettle, + Handler, + Unknown, +} + +impl FailureCategory { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Routing => "routing", + Self::Decode => "decode", + Self::Validation => "validation", + Self::Auth => "auth", + Self::Guard => "guard", + Self::Repository => "repository", + Self::Storage => "storage", + Self::Transport => "transport", + Self::OutboxPublish => "outbox_publish", + Self::OutboxSettle => "outbox_settle", + Self::Handler => "handler", + Self::Unknown => UNKNOWN, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum FailureAction { + Nack, + DeadLetter, + Park, + LogAndAck, + Stop, + Release, + Fail, + Return400, + Return401, + Return404, + Return422, + Return500, + Return503, + RecvError, + SettleAck, + SettleNack, + SettleDeadLetter, + SettlePark, +} + +impl FailureAction { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Nack => "nack", + Self::DeadLetter => "dead_letter", + Self::Park => "park", + Self::LogAndAck => "log_and_ack", + Self::Stop => "stop", + Self::Release => "release", + Self::Fail => "fail", + Self::Return400 => "return_400", + Self::Return401 => "return_401", + Self::Return404 => "return_404", + Self::Return422 => "return_422", + Self::Return500 => "return_500", + Self::Return503 => "return_503", + Self::RecvError => "recv_error", + Self::SettleAck => "settle_ack", + Self::SettleNack => "settle_nack", + Self::SettleDeadLetter => "settle_dead_letter", + Self::SettlePark => "settle_park", + } + } + + pub(crate) const fn for_status(status: u16) -> Self { + match status { + 400 => Self::Return400, + 401 => Self::Return401, + 404 => Self::Return404, + 422 => Self::Return422, + 503 => Self::Return503, + _ => Self::Return500, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[expect( + dead_code, + reason = "schema vocabulary reserves unknown when a future source cannot classify retryability" +)] +pub(crate) enum FailureRetryClass { + Retryable, + Permanent, + Unknown, +} + +impl FailureRetryClass { + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Retryable => "retryable", + Self::Permanent => "permanent", + Self::Unknown => UNKNOWN, + } + } +} + +impl From for FailureRetryClass { + fn from(kind: TransportErrorKind) -> Self { + match kind { + TransportErrorKind::Retryable => Self::Retryable, + TransportErrorKind::Permanent => Self::Permanent, + } + } +} + +impl From for FailureRetryClass { + fn from(kind: RetryClass) -> Self { + match kind { + RetryClass::Retryable => Self::Retryable, + RetryClass::Permanent => Self::Permanent, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct HandlerFailureMapping { + pub(crate) category: FailureCategory, + pub(crate) status: u16, + pub(crate) retry_class: FailureRetryClass, + pub(crate) action: FailureAction, + pub(crate) error_type: &'static str, +} + +pub(crate) fn handler_failure_mapping(error: &HandlerError) -> HandlerFailureMapping { + let status = error.status_code(); + let (category, retry_class, error_type) = match error { + HandlerError::UnknownCommand(_) => ( + FailureCategory::Routing, + FailureRetryClass::Permanent, + "HandlerError::UnknownCommand", + ), + HandlerError::DecodeFailed(_) => ( + FailureCategory::Decode, + FailureRetryClass::Permanent, + "HandlerError::DecodeFailed", + ), + HandlerError::Rejected(_) => ( + FailureCategory::Validation, + FailureRetryClass::Permanent, + "HandlerError::Rejected", + ), + HandlerError::NotFound(_) => ( + FailureCategory::Repository, + FailureRetryClass::Retryable, + "HandlerError::NotFound", + ), + HandlerError::Unauthorized(_) => ( + FailureCategory::Auth, + FailureRetryClass::Permanent, + "HandlerError::Unauthorized", + ), + HandlerError::Repository(err) => ( + repository_failure_category(err), + FailureRetryClass::from(err.kind()), + repository_error_type(err), + ), + HandlerError::GuardRejected(_) => ( + FailureCategory::Guard, + FailureRetryClass::Permanent, + "HandlerError::GuardRejected", + ), + HandlerError::Other(_) => ( + FailureCategory::Handler, + FailureRetryClass::Retryable, + "HandlerError::Other", + ), + }; + + HandlerFailureMapping { + category, + status, + retry_class, + action: FailureAction::for_status(status), + error_type, + } +} + +pub(crate) fn repository_failure_category(error: &RepositoryError) -> FailureCategory { + match error { + RepositoryError::LockPoisoned(_) + | RepositoryError::Lock(_) + | RepositoryError::Storage { .. } => FailureCategory::Storage, + RepositoryError::ConcurrentWrite { .. } + | RepositoryError::DuplicateStreamInBatch { .. } + | RepositoryError::DuplicateOutboxMessageInBatch { .. } + | RepositoryError::DuplicateInboxReceipt { .. } + | RepositoryError::InvalidInboxReceipt { .. } + | RepositoryError::InvalidStreamIdentity { .. } + | RepositoryError::NotFound { .. } + | RepositoryError::InvalidState { .. } + | RepositoryError::Replay(_) + | RepositoryError::Model(_) => FailureCategory::Repository, + } +} + +pub(crate) fn repository_error_type(error: &RepositoryError) -> &'static str { + match error { + RepositoryError::LockPoisoned(_) => "RepositoryError::LockPoisoned", + RepositoryError::Lock(_) => "RepositoryError::Lock", + RepositoryError::ConcurrentWrite { .. } => "RepositoryError::ConcurrentWrite", + RepositoryError::DuplicateStreamInBatch { .. } => "RepositoryError::DuplicateStreamInBatch", + RepositoryError::DuplicateOutboxMessageInBatch { .. } => { + "RepositoryError::DuplicateOutboxMessageInBatch" + } + RepositoryError::DuplicateInboxReceipt { .. } => "RepositoryError::DuplicateInboxReceipt", + RepositoryError::InvalidInboxReceipt { .. } => "RepositoryError::InvalidInboxReceipt", + RepositoryError::InvalidStreamIdentity { .. } => "RepositoryError::InvalidStreamIdentity", + RepositoryError::NotFound { .. } => "RepositoryError::NotFound", + RepositoryError::InvalidState { .. } => "RepositoryError::InvalidState", + RepositoryError::Replay(_) => "RepositoryError::Replay", + RepositoryError::Model(_) => "RepositoryError::Model", + RepositoryError::Storage { retryable, .. } => { + if *retryable { + "RepositoryError::StorageRetryable" + } else { + "RepositoryError::StoragePermanent" + } + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub(crate) struct FailureMessageFields { + pub(crate) kind: Option, + pub(crate) name: String, + pub(crate) id_hash: String, + pub(crate) correlation_id: String, + pub(crate) causation_id: String, + pub(crate) trace: FailureTraceFields, + pub(crate) payload_size_bytes: Option, + pub(crate) payload_content_type: String, + pub(crate) payload_codec: String, +} + +impl FailureMessageFields { + pub(crate) fn unknown() -> Self { + Self { + name: UNKNOWN.to_string(), + ..Self::default() + } + } + + #[cfg(any(feature = "http", feature = "grpc"))] + pub(crate) fn for_name(kind: MessageKind, name: &str) -> Self { + Self { + kind: Some(kind), + name: cap(name, MESSAGE_LIMIT), + ..Self::default() + } + } + + #[cfg(any(feature = "http", feature = "grpc"))] + pub(crate) fn for_name_with_metadata( + kind: MessageKind, + name: &str, + metadata: &[(String, String)], + ) -> Self { + let mut fields = Self::for_name(kind, name); + fields.correlation_id = metadata_value(metadata, crate::trace_context::CORRELATION_ID) + .map(cap_id) + .unwrap_or_default(); + fields.causation_id = metadata_value(metadata, crate::trace_context::CAUSATION_ID) + .map(cap_id) + .unwrap_or_default(); + fields.trace = trace_fields_from_metadata(metadata); + fields + } + + pub(crate) fn from_message(message: &Message) -> Self { + let trace = trace_fields_from_metadata(&message.metadata); + Self { + kind: Some(message.kind), + name: cap(message.name(), MESSAGE_LIMIT), + id_hash: message.id().map(hash_identifier).unwrap_or_default(), + correlation_id: message.correlation_id().map(cap_id).unwrap_or_default(), + causation_id: message.causation_id().map(cap_id).unwrap_or_default(), + trace, + payload_size_bytes: Some(message.payload().len()), + payload_content_type: sanitize_label(&message.content_type), + payload_codec: message + .metadata("x-sourced-payload-codec") + .map(sanitize_label) + .unwrap_or_default(), + } + } +} + +#[cfg(any(feature = "http", feature = "grpc"))] +fn metadata_value<'a>(metadata: &'a [(String, String)], key: &str) -> Option<&'a str> { + metadata + .iter() + .find(|(existing, _)| existing.eq_ignore_ascii_case(key)) + .map(|(_, value)| value.as_str()) +} + +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub(crate) struct FailureTraceFields { + pub(crate) trace_id: String, + pub(crate) parent_span_id: String, + pub(crate) trace_flags: String, + pub(crate) traceparent_valid: bool, + pub(crate) tracestate_present: bool, +} + +fn trace_fields_from_metadata(metadata: &[(String, String)]) -> FailureTraceFields { + let context = TraceContext::from_metadata(metadata.iter().map(|(key, value)| (key, value))); + let tracestate_present = context.tracestate.is_some(); + let Some(traceparent) = context.traceparent.as_deref() else { + return FailureTraceFields { + tracestate_present, + ..FailureTraceFields::default() + }; + }; + if !is_valid_traceparent(traceparent) { + return FailureTraceFields { + traceparent_valid: false, + tracestate_present, + ..FailureTraceFields::default() + }; + } + + let mut parts = traceparent.split('-'); + let _version = parts.next(); + let trace_id = parts.next().unwrap_or_default().to_string(); + let parent_span_id = parts.next().unwrap_or_default().to_string(); + let trace_flags = parts.next().unwrap_or_default().to_string(); + + FailureTraceFields { + trace_id, + parent_span_id, + trace_flags, + traceparent_valid: true, + tracestate_present, + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct FailureRecord { + pub(crate) service_name: String, + pub(crate) component: FailureComponent, + pub(crate) operation: FailureOperation, + pub(crate) category: FailureCategory, + pub(crate) action: FailureAction, + pub(crate) retry_class: FailureRetryClass, + pub(crate) message: FailureMessageFields, + pub(crate) error_type: &'static str, + pub(crate) error_message: String, + pub(crate) http_status_code: Option, + pub(crate) grpc_status_code: Option, + pub(crate) transport: String, + pub(crate) outbox_attempt: Option, + pub(crate) outbox_max_attempts: Option, + pub(crate) outbox_status: String, + pub(crate) outbox_source_aggregate_type: String, + pub(crate) outbox_source_sequence: String, +} + +impl FailureRecord { + pub(crate) fn new( + component: FailureComponent, + operation: FailureOperation, + category: FailureCategory, + action: FailureAction, + retry_class: FailureRetryClass, + error_type: &'static str, + error_message: impl AsRef, + ) -> Self { + Self { + service_name: String::new(), + component, + operation, + category, + action, + retry_class, + message: FailureMessageFields::unknown(), + error_type, + error_message: sanitize_error_message(error_message.as_ref()), + http_status_code: None, + grpc_status_code: None, + transport: String::new(), + outbox_attempt: None, + outbox_max_attempts: None, + outbox_status: String::new(), + outbox_source_aggregate_type: String::new(), + outbox_source_sequence: String::new(), + } + } + + pub(crate) fn from_handler_error( + component: FailureComponent, + operation: FailureOperation, + error: &HandlerError, + ) -> Self { + let mapping = handler_failure_mapping(error); + let error_message = handler_error_summary(error); + Self::new( + component, + operation, + mapping.category, + mapping.action, + mapping.retry_class, + mapping.error_type, + error_message, + ) + .with_http_status(mapping.status) + } + + pub(crate) fn from_transport_error( + component: FailureComponent, + operation: FailureOperation, + action: FailureAction, + error: &TransportError, + ) -> Self { + let (category, error_type) = classify_transport_error(error); + Self::new( + component, + operation, + category, + action, + FailureRetryClass::from(error.kind()), + error_type, + error.message(), + ) + } + + pub(crate) fn from_repository_error( + component: FailureComponent, + operation: FailureOperation, + action: FailureAction, + error: &RepositoryError, + ) -> Self { + Self::new( + component, + operation, + repository_failure_category(error), + action, + FailureRetryClass::from(error.kind()), + repository_error_type(error), + error.to_string(), + ) + } + + pub(crate) fn with_service(mut self, service_name: Option<&str>) -> Self { + self.service_name = service_name + .map(|name| cap(name, TYPE_LIMIT)) + .unwrap_or_default(); + self + } + + pub(crate) fn with_message(mut self, message: FailureMessageFields) -> Self { + self.message = message; + self + } + + pub(crate) fn with_http_status(mut self, status: u16) -> Self { + self.http_status_code = Some(status); + self + } + + #[cfg(feature = "http")] + pub(crate) fn with_action(mut self, action: FailureAction) -> Self { + self.action = action; + self + } + + pub(crate) fn with_category(mut self, category: FailureCategory) -> Self { + self.category = category; + self + } + + #[cfg(feature = "grpc")] + pub(crate) fn with_grpc_status(mut self, status: u16) -> Self { + self.grpc_status_code = Some(status); + self + } + + pub(crate) fn with_transport(mut self, transport: &str) -> Self { + self.transport = cap(transport, TYPE_LIMIT); + self + } + + pub(crate) fn with_outbox_fields( + mut self, + attempt: u32, + max_attempts: u32, + status: &'static str, + source_aggregate_type: String, + source_sequence: String, + ) -> Self { + self.outbox_attempt = Some(attempt); + self.outbox_max_attempts = Some(max_attempts); + self.outbox_status = status.to_string(); + self.outbox_source_aggregate_type = sanitize_label(&source_aggregate_type); + self.outbox_source_sequence = sanitize_label(&source_sequence); + self + } + + pub(crate) fn emit(&self) { + emit_failure(self); + } +} + +fn classify_transport_error(error: &TransportError) -> (FailureCategory, &'static str) { + if let Some(source) = error.source() { + if let Some(handler) = source.downcast_ref::() { + let mapping = handler_failure_mapping(handler); + return (mapping.category, mapping.error_type); + } + if let Some(repository) = source.downcast_ref::() { + return ( + repository_failure_category(repository), + repository_error_type(repository), + ); + } + } + + let error_type = match error.kind() { + TransportErrorKind::Retryable => "TransportError::Retryable", + TransportErrorKind::Permanent => "TransportError::Permanent", + }; + (FailureCategory::Transport, error_type) +} + +fn handler_error_summary(error: &HandlerError) -> String { + match error { + HandlerError::UnknownCommand(_) => "unknown command".to_string(), + HandlerError::DecodeFailed(message) => format!("decode failed: {message}"), + HandlerError::Rejected(message) => format!("rejected: {message}"), + HandlerError::NotFound(_) => "not found".to_string(), + HandlerError::Unauthorized(_) => "unauthorized".to_string(), + HandlerError::Repository(error) => error.to_string(), + HandlerError::GuardRejected(_) => "guard rejected command".to_string(), + HandlerError::Other(error) => error.to_string(), + } +} + +pub(crate) fn sanitize_error_message(value: &str) -> String { + let lower = value.to_ascii_lowercase(); + if looks_like_sql(&lower) { + return "sql statement redacted".to_string(); + } + + let mut redacted = Vec::new(); + let mut redact_next = false; + for token in value.split_whitespace() { + let lower = token.to_ascii_lowercase(); + let sensitive = redact_next || is_sensitive_token(&lower); + if sensitive { + redacted.push("[redacted]".to_string()); + redact_next = redacts_following_value(&lower); + } else if lower.contains("://") { + redacted.push("[redacted-url]".to_string()); + redact_next = false; + } else { + redacted.push(token.to_string()); + redact_next = false; + } + } + + cap(&redacted.join(" "), ERROR_LIMIT) +} + +fn looks_like_sql(lower: &str) -> bool { + lower.contains("select ") + || lower.contains("insert ") + || lower.contains("update ") + || lower.contains("delete ") + || lower.contains(" where ") + || lower.contains(" from ") +} + +fn is_sensitive_token(lower: &str) -> bool { + lower.contains("authorization") + || lower == "bearer" + || lower.contains("cookie") + || lower.contains("token") + || lower.contains("secret") + || lower.contains("password") + || lower.contains("passwd") + || lower.contains("api_key") + || lower.contains("apikey") + || lower.contains("access_key") + || lower.contains("session_variables") + || lower.contains("session:") + || lower.contains("x-hasura-") + || lower.contains("://") +} + +fn redacts_following_value(lower: &str) -> bool { + lower.ends_with(':') + || lower.ends_with('=') + || matches!( + lower, + "authorization" + | "bearer" + | "cookie" + | "token" + | "secret" + | "password" + | "passwd" + | "api_key" + | "apikey" + | "access_key" + | "session" + | "session_variables" + ) +} + +fn cap_id(value: &str) -> String { + cap(value, ID_LIMIT) +} + +fn sanitize_label(value: &str) -> String { + cap(&sanitize_error_message(value), TYPE_LIMIT) +} + +fn cap(value: &str, max: usize) -> String { + value.chars().take(max).collect() +} + +fn hash_identifier(value: &str) -> String { + const OFFSET: u64 = 0xcbf29ce484222325; + const PRIME: u64 = 0x00000100000001B3; + + let mut hash = OFFSET; + for byte in value.as_bytes() { + hash ^= u64::from(*byte); + hash = hash.wrapping_mul(PRIME); + } + format!("{hash:016x}") +} + +#[cfg(feature = "failure-logs")] +fn emit_failure(record: &FailureRecord) { + let message_kind = record.message.kind.map(|kind| kind.as_str()).unwrap_or(""); + let payload_size = record.message.payload_size_bytes.unwrap_or_default(); + let http_status = record.http_status_code.unwrap_or_default(); + let grpc_status = record.grpc_status_code.unwrap_or_default(); + let outbox_attempt = record.outbox_attempt.unwrap_or_default(); + let outbox_max_attempts = record.outbox_max_attempts.unwrap_or_default(); + + tracing::event!( + target: "distributed.failure", + tracing::Level::ERROR, + event.name = "distributed.failure", + distributed.failure.schema_version = SCHEMA_VERSION, + distributed.service.name = %record.service_name, + distributed.component = record.component.as_str(), + distributed.operation = record.operation.as_str(), + distributed.message.kind = message_kind, + distributed.message.name = %record.message.name, + messaging.message.id_hash = %record.message.id_hash, + distributed.correlation_id = %record.message.correlation_id, + distributed.causation_id = %record.message.causation_id, + trace.trace_id = %record.message.trace.trace_id, + trace.parent_span_id = %record.message.trace.parent_span_id, + trace.trace_flags = %record.message.trace.trace_flags, + trace.traceparent_valid = record.message.trace.traceparent_valid, + trace.tracestate_present = record.message.trace.tracestate_present, + error.type = record.error_type, + error.category = record.category.as_str(), + error.retry_class = record.retry_class.as_str(), + error.message = %record.error_message, + distributed.failure.action = record.action.as_str(), + http.response.status_code = http_status, + rpc.grpc.status_code = grpc_status, + messaging.system = %record.transport, + outbox.attempt = outbox_attempt, + outbox.max_attempts = outbox_max_attempts, + outbox.status = %record.outbox_status, + outbox.source_aggregate_type = %record.outbox_source_aggregate_type, + outbox.source_sequence = %record.outbox_source_sequence, + payload.size_bytes = payload_size, + payload.content_type = %record.message.payload_content_type, + payload.codec = %record.message.payload_codec, + ); +} + +#[cfg(not(feature = "failure-logs"))] +fn emit_failure(record: &FailureRecord) { + let _ = ( + SCHEMA_VERSION, + record.component.as_str(), + record.operation.as_str(), + record.category.as_str(), + record.retry_class.as_str(), + ); +} + +#[cfg(all(test, feature = "failure-logs"))] +pub(crate) mod testing { + use std::collections::BTreeMap; + use std::sync::{Arc, Mutex, MutexGuard, OnceLock}; + + use tracing::field::{Field, Visit}; + use tracing::{Event, Subscriber}; + use tracing_subscriber::layer::Context; + use tracing_subscriber::prelude::*; + use tracing_subscriber::{Layer, Registry}; + + static EVENTS: OnceLock>>> = OnceLock::new(); + static TEST_LOCK: Mutex<()> = Mutex::new(()); + + #[derive(Debug, Clone, Default, PartialEq, Eq)] + pub(crate) struct CapturedEvent { + pub(crate) target: String, + pub(crate) fields: BTreeMap, + } + + impl CapturedEvent { + pub(crate) fn field(&self, name: &str) -> Option<&str> { + self.fields.get(name).map(String::as_str) + } + } + + pub(crate) struct CaptureGuard { + _guard: MutexGuard<'static, ()>, + events: Arc>>, + } + + impl CaptureGuard { + pub(crate) fn events(&self) -> Vec { + self.events + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + } + + pub(crate) fn failure_events(&self) -> Vec { + self.events() + .into_iter() + .filter(|event| event.target == "distributed.failure") + .collect() + } + } + + pub(crate) fn capture_failures() -> CaptureGuard { + let events = EVENTS + .get_or_init(|| { + let events = Arc::new(Mutex::new(Vec::new())); + let subscriber = Registry::default().with(FailureCaptureLayer { + events: events.clone(), + }); + let _ = tracing::subscriber::set_global_default(subscriber); + events + }) + .clone(); + + let guard = TEST_LOCK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + events + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clear(); + CaptureGuard { + _guard: guard, + events, + } + } + + struct FailureCaptureLayer { + events: Arc>>, + } + + impl Layer for FailureCaptureLayer + where + S: Subscriber, + { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + if event.metadata().target() != "distributed.failure" { + return; + } + let mut visitor = FieldVisitor::default(); + event.record(&mut visitor); + self.events + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .push(CapturedEvent { + target: event.metadata().target().to_string(), + fields: visitor.fields, + }); + } + } + + #[derive(Default)] + struct FieldVisitor { + fields: BTreeMap, + } + + impl Visit for FieldVisitor { + fn record_bool(&mut self, field: &Field, value: bool) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } + + fn record_i64(&mut self, field: &Field, value: i64) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } + + fn record_u64(&mut self, field: &Field, value: u64) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } + + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.fields + .insert(field.name().to_string(), format!("{value:?}")); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const TRACEPARENT: &str = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + + #[test] + fn handler_error_mapping_covers_all_variants() { + let cases = [ + ( + HandlerError::UnknownCommand("raw-path".into()), + FailureCategory::Routing, + 404, + FailureRetryClass::Permanent, + FailureAction::Return404, + ), + ( + HandlerError::DecodeFailed("bad json".into()), + FailureCategory::Decode, + 400, + FailureRetryClass::Permanent, + FailureAction::Return400, + ), + ( + HandlerError::Rejected("invalid".into()), + FailureCategory::Validation, + 422, + FailureRetryClass::Permanent, + FailureAction::Return422, + ), + ( + HandlerError::NotFound("aggregate-1".into()), + FailureCategory::Repository, + 404, + FailureRetryClass::Retryable, + FailureAction::Return404, + ), + ( + HandlerError::Unauthorized("missing user".into()), + FailureCategory::Auth, + 401, + FailureRetryClass::Permanent, + FailureAction::Return401, + ), + ( + HandlerError::Repository(RepositoryError::Model("bad row".into())), + FailureCategory::Repository, + 500, + FailureRetryClass::Permanent, + FailureAction::Return500, + ), + ( + HandlerError::GuardRejected("order.create".into()), + FailureCategory::Guard, + 400, + FailureRetryClass::Permanent, + FailureAction::Return400, + ), + ( + HandlerError::Other(Box::::from("io failed")), + FailureCategory::Handler, + 500, + FailureRetryClass::Retryable, + FailureAction::Return500, + ), + ]; + + for (error, category, status, retry_class, action) in cases { + let mapping = handler_failure_mapping(&error); + assert_eq!( + ( + mapping.category, + mapping.status, + mapping.retry_class, + mapping.action + ), + (category, status, retry_class, action), + "{error}" + ); + } + } + + #[test] + fn repository_and_transport_retry_classification_is_reused() { + let retryable_repo = RepositoryError::retryable_storage( + "load stream", + std::io::Error::new(std::io::ErrorKind::TimedOut, "timed out"), + ); + let permanent_repo = RepositoryError::permanent_storage( + "insert row", + std::io::Error::new(std::io::ErrorKind::InvalidData, "constraint"), + ); + assert_eq!( + FailureRetryClass::from(retryable_repo.kind()), + FailureRetryClass::Retryable + ); + assert_eq!( + FailureRetryClass::from(permanent_repo.kind()), + FailureRetryClass::Permanent + ); + + let retryable = TransportError::retryable("broker timed out"); + let permanent = TransportError::permanent("decode failed"); + assert_eq!( + FailureRecord::from_transport_error( + FailureComponent::Transport, + FailureOperation::Receive, + FailureAction::RecvError, + &retryable, + ) + .retry_class, + FailureRetryClass::Retryable + ); + assert_eq!( + FailureRecord::from_transport_error( + FailureComponent::Transport, + FailureOperation::Receive, + FailureAction::RecvError, + &permanent, + ) + .retry_class, + FailureRetryClass::Permanent + ); + } + + #[test] + fn sanitizer_removes_sensitive_values_and_db_urls() { + let raw = "authorization: Bearer abc123 cookie: session=xyz password hunter2 token tok secret=s postgres://user:pass@db/app"; + + let sanitized = sanitize_error_message(raw); + + for forbidden in [ + "abc123", + "session=xyz", + "hunter2", + "tok", + "secret=s", + "postgres://user:pass@db/app", + ] { + assert!( + !sanitized.contains(forbidden), + "sanitized message leaked `{forbidden}`: {sanitized}" + ); + } + } + + #[test] + fn sanitizer_removes_raw_sql() { + let sanitized = + sanitize_error_message("SELECT * FROM users WHERE password = 'raw-password'"); + + assert_eq!(sanitized, "sql statement redacted"); + } + + #[test] + fn record_excludes_payload_metadata_session_values_and_raw_tracestate() { + let mut message = Message::new( + "orders.created", + MessageKind::Event, + br#"{"request_body":"payload-secret"}"#.to_vec(), + ) + .with_id("tenant-user-message-id") + .with_metadata("authorization", "Bearer raw-auth") + .with_metadata("cookie", "raw-cookie") + .with_metadata("x-hasura-user-id", "user-42") + .with_metadata("arbitrary", "metadata-secret") + .with_metadata("x-sourced-payload-codec", "metadata-secret") + .with_metadata("correlation_id", "corr-1") + .with_metadata("causation_id", "cause-1") + .with_metadata("traceparent", TRACEPARENT) + .with_metadata("tracestate", "vendor=raw-state"); + message.content_type = "application/json; password=hunter2".to_string(); + let error = HandlerError::Repository(RepositoryError::Model( + "failed with token=raw-token password=raw-password".into(), + )); + + let record = FailureRecord::from_handler_error( + FailureComponent::Microsvc, + FailureOperation::Dispatch, + &error, + ) + .with_message(FailureMessageFields::from_message(&message)); + let rendered = format!("{record:?}"); + + for forbidden in [ + "payload-secret", + "raw-auth", + "raw-cookie", + "user-42", + "metadata-secret", + "raw-state", + "raw-token", + "raw-password", + "hunter2", + "tenant-user-message-id", + ] { + assert!( + !rendered.contains(forbidden), + "failure record leaked `{forbidden}`: {rendered}" + ); + } + assert!(rendered.contains("corr-1")); + assert!(rendered.contains("cause-1")); + assert!(record.message.trace.tracestate_present); + assert_eq!( + record.message.trace.trace_id, + "4bf92f3577b34da6a3ce929d0e0e4736" + ); + } +} diff --git a/src/lib.rs b/src/lib.rs index d6a0112..7831da7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -12,6 +12,7 @@ pub mod repository; mod commit_builder; #[cfg(feature = "emitter")] pub mod emitter; +mod failure_log; mod in_memory_repo; pub mod lock; pub mod manifest; diff --git a/src/microsvc/grpc.rs b/src/microsvc/grpc.rs index e612250..e24dcc5 100644 --- a/src/microsvc/grpc.rs +++ b/src/microsvc/grpc.rs @@ -41,6 +41,11 @@ use tonic::{Request, Response, Status}; use super::service::Service; use super::session::Session; +use crate::bus::MessageKind; +use crate::failure_log::{ + FailureAction, FailureCategory, FailureComponent, FailureMessageFields, FailureOperation, + FailureRecord, FailureRetryClass, +}; // --------------------------------------------------------------------------- // Message types (prost — standard protobuf wire format) @@ -156,6 +161,22 @@ impl CommandService for GrpcHandler { let input: serde_json::Value = match serde_json::from_str(&req.input) { Ok(value) => value, Err(e) => { + FailureRecord::new( + FailureComponent::Grpc, + FailureOperation::Ingress, + FailureCategory::Decode, + FailureAction::Return400, + FailureRetryClass::Permanent, + "serde_json::Error", + format!("invalid JSON input: {e}"), + ) + .with_service(self.service.name()) + .with_message(FailureMessageFields::for_name( + MessageKind::Command, + &req.command, + )) + .with_grpc_status(400) + .emit(); return Ok(Response::new(GrpcResponse { status: 400, body: json!({ "error": format!("invalid JSON input: {e}") }).to_string(), @@ -168,6 +189,7 @@ impl CommandService for GrpcHandler { // payload is client-controlled and MUST NOT override it. See // [`build_session`] and the `Session` trust-boundary docs. let session = build_session(&metadata, req.session_variables); + let failure_metadata = session_metadata(&session); match self.service.dispatch(&req.command, input, session).await { Ok(value) => Ok(Response::new(GrpcResponse { @@ -179,9 +201,24 @@ impl CommandService for GrpcHandler { // (SQL/driver text) to clients. Server-internal failures are // masked; client-fault errors keep their descriptive message. let status = e.status_code(); - if status >= 500 { - eprintln!("microsvc command `{}` failed: {e}", req.command); - } + let message = if matches!(e, super::error::HandlerError::UnknownCommand(_)) { + FailureMessageFields::unknown() + } else { + FailureMessageFields::for_name_with_metadata( + MessageKind::Command, + &req.command, + &failure_metadata, + ) + }; + FailureRecord::from_handler_error( + FailureComponent::Grpc, + FailureOperation::Ingress, + &e, + ) + .with_service(self.service.name()) + .with_message(message) + .with_grpc_status(status) + .emit(); Ok(Response::new(GrpcResponse { status: status as u32, body: json!({ "error": e.client_facing_message() }).to_string(), @@ -248,6 +285,14 @@ fn build_session( Session::from_map(vars) } +fn session_metadata(session: &Session) -> Vec<(String, String)> { + session + .variables() + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect() +} + // --------------------------------------------------------------------------- // Convenience constructors // --------------------------------------------------------------------------- @@ -275,3 +320,82 @@ pub async fn serve_grpc(service: Arc, addr: &str) -> Result<(), GrpcSer .await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[cfg(feature = "failure-logs")] + #[tokio::test] + async fn dispatch_masks_5xx_and_emits_sanitized_failure_event() { + use crate::microsvc::{Context, HandlerError, Routes}; + use crate::RepositoryError; + + let service = Arc::new( + Service::new().named("orders-grpc").routes( + Routes::new() + .with_dependencies(()) + .command("orders.create") + .handle(|_: &Context<()>| async move { + Err(HandlerError::Repository(RepositoryError::Model( + "storage failed token=raw-token mysql://user:pass@db/app".into(), + ))) + }), + ), + ); + let handler = GrpcHandler::new(service); + let mut payload_vars = HashMap::new(); + payload_vars.insert("x-hasura-user-id".to_string(), "user-secret".to_string()); + let mut request = Request::new(GrpcRequest { + command: "orders.create".to_string(), + input: json!({ "request_body": "payload-secret" }).to_string(), + session_variables: payload_vars, + }); + request + .metadata_mut() + .insert("authorization", "Bearer raw-auth".parse().unwrap()); + request.metadata_mut().insert( + "traceparent", + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" + .parse() + .unwrap(), + ); + request + .metadata_mut() + .insert("tracestate", "vendor=raw-state".parse().unwrap()); + let capture = crate::failure_log::testing::capture_failures(); + + let response = handler.dispatch(request).await.unwrap().into_inner(); + + assert_eq!(response.status, 500); + let body: serde_json::Value = serde_json::from_str(&response.body).unwrap(); + assert_eq!(body, json!({ "error": "Internal server error" })); + + let events = capture.failure_events(); + let grpc = events + .iter() + .find(|event| event.field("distributed.component") == Some("grpc")) + .expect("grpc failure event"); + assert_eq!(grpc.field("distributed.failure.action"), Some("return_500")); + assert_eq!(grpc.field("rpc.grpc.status_code"), Some("500")); + assert_eq!( + grpc.field("trace.trace_id"), + Some("4bf92f3577b34da6a3ce929d0e0e4736") + ); + + let rendered = format!("{events:?}"); + for forbidden in [ + "payload-secret", + "raw-token", + "mysql://user:pass@db/app", + "user-secret", + "raw-auth", + "raw-state", + ] { + assert!( + !rendered.contains(forbidden), + "failure event leaked `{forbidden}`: {rendered}" + ); + } + } +} diff --git a/src/microsvc/http.rs b/src/microsvc/http.rs index 68bf25b..169b78e 100644 --- a/src/microsvc/http.rs +++ b/src/microsvc/http.rs @@ -44,6 +44,8 @@ use super::error::HandlerError; use super::service::Service; use super::session::Session; use super::MAX_HTTP_BODY_BYTES; +use crate::bus::MessageKind; +use crate::failure_log::{FailureComponent, FailureMessageFields, FailureOperation, FailureRecord}; /// Build an axum `Router` that dispatches commands via the given service. pub fn router(service: Arc) -> Router { @@ -91,13 +93,12 @@ async fn command_handler( Json(input): Json, ) -> impl IntoResponse { let session = session_from_headers(&headers); + let metadata = session_metadata(&session); match service.dispatch(&command, input, session).await { Ok(value) => (StatusCode::OK, Json(value)).into_response(), Err(err) => { let status = status_for_error(&err); - if status.is_server_error() { - eprintln!("microsvc command `{command}` failed: {err}"); - } + emit_http_failure(service.as_ref(), &command, &metadata, status, &err); let body = json!({ "error": err.client_facing_message() }); (status, Json(body)).into_response() } @@ -114,6 +115,25 @@ fn status_for_error(error: &HandlerError) -> StatusCode { } } +fn emit_http_failure( + service: &Service, + command: &str, + metadata: &[(String, String)], + status: StatusCode, + error: &HandlerError, +) { + let message = if matches!(error, HandlerError::UnknownCommand(_)) { + FailureMessageFields::unknown() + } else { + FailureMessageFields::for_name_with_metadata(MessageKind::Command, command, metadata) + }; + FailureRecord::from_handler_error(FailureComponent::Http, FailureOperation::Ingress, error) + .with_service(service.name()) + .with_message(message) + .with_http_status(status.as_u16()) + .emit(); +} + /// Extract session variables from HTTP headers. /// /// **Trust boundary (security-critical):** every request header is copied @@ -132,6 +152,14 @@ fn session_from_headers(headers: &HeaderMap) -> Session { Session::from_map(vars) } +fn session_metadata(session: &Session) -> Vec<(String, String)> { + session + .variables() + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect() +} + #[cfg(test)] mod tests { use super::*; @@ -200,4 +228,80 @@ mod tests { assert_eq!(error.client_facing_message(), "Internal server error"); } } + + #[cfg(feature = "failure-logs")] + #[tokio::test] + async fn command_handler_masks_5xx_and_emits_sanitized_failure_event() { + use crate::microsvc::Routes; + use axum::body; + use axum::http::HeaderValue; + + let service = Arc::new( + Service::new().named("orders").routes( + Routes::new() + .with_dependencies(()) + .command("orders.create") + .handle(|_: &crate::microsvc::Context<()>| async move { + Err(HandlerError::Repository(RepositoryError::Model( + "database failed password=hunter2 postgres://user:pass@db/app".into(), + ))) + }), + ), + ); + let mut headers = HeaderMap::new(); + headers.insert( + "authorization", + HeaderValue::from_static("Bearer raw-token"), + ); + headers.insert("cookie", HeaderValue::from_static("session=raw-cookie")); + headers.insert( + "traceparent", + HeaderValue::from_static("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"), + ); + headers.insert("tracestate", HeaderValue::from_static("vendor=raw-state")); + let capture = crate::failure_log::testing::capture_failures(); + + let response = command_handler( + State(service), + Path("orders.create".to_string()), + headers, + Json(json!({ "request_body": "payload-secret" })), + ) + .await + .into_response(); + + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + let body = body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body, json!({ "error": "Internal server error" })); + + let events = capture.failure_events(); + let http = events + .iter() + .find(|event| event.field("distributed.component") == Some("http")) + .expect("http failure event"); + assert_eq!(http.field("distributed.failure.action"), Some("return_500")); + assert_eq!(http.field("error.category"), Some("repository")); + assert_eq!( + http.field("trace.trace_id"), + Some("4bf92f3577b34da6a3ce929d0e0e4736") + ); + + let rendered = format!("{events:?}"); + for forbidden in [ + "payload-secret", + "raw-token", + "raw-cookie", + "raw-state", + "hunter2", + "postgres://user:pass@db/app", + ] { + assert!( + !rendered.contains(forbidden), + "failure event leaked `{forbidden}`: {rendered}" + ); + } + } } diff --git a/src/microsvc/knative_ingress.rs b/src/microsvc/knative_ingress.rs index 292ced5..4611672 100644 --- a/src/microsvc/knative_ingress.rs +++ b/src/microsvc/knative_ingress.rs @@ -32,6 +32,10 @@ use base64::Engine; use serde_json::{json, Value}; use crate::bus::{validate_message_name, Message, MessageKind}; +use crate::failure_log::{ + FailureAction, FailureCategory, FailureComponent, FailureMessageFields, FailureOperation, + FailureRecord, FailureRetryClass, +}; use crate::microsvc::{Service, MAX_HTTP_BODY_BYTES}; use crate::trace_context::{TraceContext, TRACEPARENT, TRACESTATE}; @@ -69,7 +73,21 @@ async fn ingress_handler( ) -> Response { let message = match parse_cloud_event(&headers, &body) { Ok(message) => message, - Err(reason) => return (StatusCode::BAD_REQUEST, reason).into_response(), + Err(reason) => { + FailureRecord::new( + FailureComponent::Knative, + FailureOperation::Ingress, + FailureCategory::Decode, + FailureAction::Return400, + FailureRetryClass::Permanent, + "CloudEventDecodeError", + &reason, + ) + .with_service(service.name()) + .with_http_status(StatusCode::BAD_REQUEST.as_u16()) + .emit(); + return (StatusCode::BAD_REQUEST, reason).into_response(); + } }; match service.dispatch_message(&message).await { @@ -82,12 +100,16 @@ async fn ingress_handler( } else { StatusCode::UNPROCESSABLE_ENTITY }; - // Mask internal faults (the same redaction the HTTP ingress applies): - // `err.to_string()` can carry SQL/driver/path detail. Log the real - // error server-side; return only a safe message on the wire. - if err.status_code() >= 500 { - eprintln!("knative ingress `{}` failed: {err}", message.name()); - } + FailureRecord::from_handler_error( + FailureComponent::Knative, + FailureOperation::Ingress, + &err, + ) + .with_service(service.name()) + .with_message(FailureMessageFields::from_message(&message)) + .with_action(FailureAction::for_status(status.as_u16())) + .with_http_status(status.as_u16()) + .emit(); ( status, Json(json!({ "error": err.client_facing_message() })), @@ -407,4 +429,77 @@ mod tests { let err = parse_cloud_event(&h, &body).unwrap_err(); assert!(err.contains("invalid cloudevent type"), "got {err}"); } + + #[cfg(feature = "failure-logs")] + #[tokio::test] + async fn ingress_masks_retryable_5xx_and_emits_sanitized_failure_event() { + let service = Arc::new( + Service::new().named("orders-knative").routes( + crate::microsvc::Routes::new() + .with_dependencies(()) + .event("order.temporarily_failed") + .handle(|_: &crate::microsvc::Context<()>| async move { + let io = std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "connection refused password=hunter2 postgres://user:pass@db/app", + ); + Err(crate::microsvc::HandlerError::Repository( + crate::RepositoryError::retryable_storage("load stream", io), + )) + }), + ), + ); + let h = headers(&[ + ("ce-id", "evt-secret-id"), + ("ce-type", "order.temporarily_failed"), + ("ce-source", "/orders"), + ( + "traceparent", + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + ), + ("tracestate", "vendor=raw-state"), + ("content-type", "application/json"), + ]); + let capture = crate::failure_log::testing::capture_failures(); + + let response = ingress_handler( + State(service), + h, + Bytes::from_static(br#"{"request_body":"payload-secret"}"#), + ) + .await; + + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body, json!({ "error": "Internal server error" })); + + let events = capture.failure_events(); + let knative = events + .iter() + .find(|event| event.field("distributed.component") == Some("knative")) + .expect("knative failure event"); + assert_eq!( + knative.field("distributed.failure.action"), + Some("return_503") + ); + assert_eq!(knative.field("http.response.status_code"), Some("503")); + assert_eq!(knative.field("error.retry_class"), Some("retryable")); + + let rendered = format!("{events:?}"); + for forbidden in [ + "payload-secret", + "hunter2", + "postgres://user:pass@db/app", + "raw-state", + "evt-secret-id", + ] { + assert!( + !rendered.contains(forbidden), + "failure event leaked `{forbidden}`: {rendered}" + ); + } + } } diff --git a/src/microsvc/service.rs b/src/microsvc/service.rs index 6efea9a..36adb27 100644 --- a/src/microsvc/service.rs +++ b/src/microsvc/service.rs @@ -47,6 +47,7 @@ use super::session::Session; use crate::bus::{ Bus, Message, MessageKind, MessagePublisher, RunOptions, SubscriptionPlan, TransportError, }; +use crate::failure_log::{FailureComponent, FailureMessageFields, FailureOperation, FailureRecord}; use crate::outbox::OutboxPublisherConfig; use crate::outbox_worker::BusOutboxPublishHook; @@ -596,7 +597,13 @@ impl Service { session: Session, ) -> Result { if !self.handles_message(MessageKind::Command, command) { - return Err(HandlerError::UnknownCommand(command.to_string())); + let error = HandlerError::UnknownCommand(command.to_string()); + self.emit_handler_failure( + FailureOperation::Dispatch, + FailureMessageFields::unknown(), + &error, + ); + return Err(error); } let payload = serde_json::to_vec(&input).map_err(|e| { @@ -616,8 +623,17 @@ impl Service { metadata, }; - self.invoke_with_dispatch_span(&message, input, session) - .await + let result = self + .invoke_with_dispatch_span(&message, input, session) + .await; + if let Err(error) = &result { + self.emit_handler_failure( + FailureOperation::Dispatch, + FailureMessageFields::from_message(&message), + error, + ); + } + result } /// Dispatch a `CommandRequest`, returning a `CommandResponse`. @@ -633,7 +649,7 @@ impl Service { }, Err(e) => CommandResponse { status: e.status_code(), - body: serde_json::json!({ "error": e.to_string() }), + body: serde_json::json!({ "error": e.client_facing_message() }), }, } } @@ -661,7 +677,13 @@ impl Service { async fn dispatch_message_inner(&self, message: &Message) -> Result { if !self.handles_message(message.kind, &message.name) { - return Err(HandlerError::UnknownCommand(message.name.clone())); + let error = HandlerError::UnknownCommand(message.name.clone()); + self.emit_handler_failure( + FailureOperation::Dispatch, + FailureMessageFields::unknown(), + &error, + ); + return Err(error); } let input = match message_to_json_input(message) { @@ -672,11 +694,27 @@ impl Service { // *claims* to be JSON but does not parse is a decode error — surface // it instead of silently nulling the input. Err(_) if !is_json_content_type(&message.content_type) => Value::Null, - Err(err) => return Err(err), + Err(err) => { + self.emit_handler_failure( + FailureOperation::Dispatch, + FailureMessageFields::from_message(message), + &err, + ); + return Err(err); + } }; let session = message_to_session(message); - self.invoke_with_dispatch_span(message, input, session) - .await + let result = self + .invoke_with_dispatch_span(message, input, session) + .await; + if let Err(error) = &result { + self.emit_handler_failure( + FailureOperation::Dispatch, + FailureMessageFields::from_message(message), + error, + ); + } + result } async fn invoke_with_dispatch_span( @@ -804,6 +842,18 @@ impl Service { ); } } + + fn emit_handler_failure( + &self, + operation: FailureOperation, + message: FailureMessageFields, + error: &HandlerError, + ) { + FailureRecord::from_handler_error(FailureComponent::Microsvc, operation, error) + .with_service(self.name()) + .with_message(message) + .emit(); + } } impl Default for Service { diff --git a/src/outbox_worker/outbox_dispatch.rs b/src/outbox_worker/outbox_dispatch.rs index 89ee6ec..cdefb2c 100644 --- a/src/outbox_worker/outbox_dispatch.rs +++ b/src/outbox_worker/outbox_dispatch.rs @@ -19,16 +19,24 @@ use futures_util::stream::{self, StreamExt}; use super::{ClaimOutboxMessages, OutboxClaimRef, OutboxPublishFailureAction, OutboxStore}; use crate::bus::{Message, MessageKind, MessagePublisher, TransportError, TransportErrorKind}; +use crate::failure_log::{ + FailureAction, FailureCategory, FailureComponent, FailureMessageFields, FailureOperation, + FailureRecord, +}; use crate::outbox::OutboxMessage; use crate::repository::RepositoryError; -/// Repository/store failures (lock contention, storage hiccups, stale-claim -/// conflicts) are retryable: usually transient, resolved by a later re-claim. This -/// conversion lives with the outbox bridge — which legitimately knows both the -/// store and the bus — so bus core stays free of `RepositoryError`. +/// Convert repository/store failures into the transport vocabulary while +/// preserving the repository retry classification. This conversion lives with +/// the outbox bridge — which legitimately knows both the store and the bus — so +/// bus core stays free of `RepositoryError`. impl From for TransportError { fn from(error: RepositoryError) -> Self { - TransportError::new(TransportErrorKind::Retryable, error.to_string()).with_source(error) + let kind = match error.kind() { + crate::lock::RetryClass::Retryable => TransportErrorKind::Retryable, + crate::lock::RetryClass::Permanent => TransportErrorKind::Permanent, + }; + TransportError::new(kind, error.to_string()).with_source(error) } } @@ -219,7 +227,19 @@ where ) -> Result { let request = ClaimOutboxMessages::for_ids(self.worker_id.clone(), ids.to_vec(), self.lease); - let claimed = self.store.claim(request).await?; + let claimed = match self.store.claim(request).await { + Ok(claimed) => claimed, + Err(error) => { + emit_outbox_repository_failure( + self.service_name.as_deref(), + FailureOperation::Claim, + FailureAction::Stop, + &error, + None, + ); + return Err(error.into()); + } + }; let mut outcome = self.dispatch_claimed(claimed).await?; outcome.requested = ids.len(); self.record_outbox_backlog().await; @@ -232,7 +252,19 @@ where batch_size: usize, ) -> Result { let request = ClaimOutboxMessages::new(self.worker_id.clone(), batch_size, self.lease); - let claimed = self.store.claim(request).await?; + let claimed = match self.store.claim(request).await { + Ok(claimed) => claimed, + Err(error) => { + emit_outbox_repository_failure( + self.service_name.as_deref(), + FailureOperation::Claim, + FailureAction::Stop, + &error, + None, + ); + return Err(error.into()); + } + }; let mut outcome = self.dispatch_claimed(claimed).await?; outcome.requested = batch_size; self.record_outbox_backlog().await; @@ -252,6 +284,7 @@ where &self.publisher, claimed, self.max_attempts, + self.service_name.as_deref(), self.publish_concurrency, ) .await?; @@ -354,6 +387,7 @@ pub(crate) async fn publish_and_settle( publisher: &P, claimed: Vec, max_attempts: u32, + service_name: Option<&str>, publish_concurrency: NonZeroUsize, ) -> Result where @@ -363,44 +397,205 @@ where let mut work = Vec::with_capacity(claimed.len()); for message in claimed { let claim = OutboxClaimRef::from_message(&message)?; - work.push((claim, Message::from(message))); + let message = Message::from(message); + let context = OutboxFailureContext::from_message(&claim, max_attempts, &message); + work.push((claim, context, message)); } // Publish phase: up to `publish_concurrency` publishes in flight. With the // default of 1 this awaits each publish before starting the next, so rows // go out strictly in claim (created-at) order. - let results: Vec<(OutboxClaimRef, Result<(), TransportError>)> = - stream::iter(work.into_iter().map(|(claim, message)| async move { - let result = publish_with_span(publisher, message).await; - (claim, result) - })) - .buffer_unordered(publish_concurrency.get()) - .collect() - .await; + let results: Vec<( + OutboxClaimRef, + OutboxFailureContext, + Result<(), TransportError>, + )> = stream::iter( + work.into_iter() + .map(|(claim, context, message)| async move { + let result = publish_with_span(publisher, message).await; + (claim, context, result) + }), + ) + .buffer_unordered(publish_concurrency.get()) + .collect() + .await; // Settle phase: failures are the exception and settle individually; // successes settle in one batched complete. let mut outcome = SettleOutcome::default(); let mut published = Vec::with_capacity(results.len()); - for (claim, result) in results { + let mut first_published_context = None; + for (claim, context, result) in results { match result { - Ok(()) => published.push(claim), + Ok(()) => { + if first_published_context.is_none() { + first_published_context = Some(context); + } + published.push(claim); + } Err(publish_error) => { + let intended_action = context.action_for_attempt(); match store .record_failure(&claim, &publish_error.to_string(), max_attempts) - .await? + .await { - OutboxPublishFailureAction::Released => outcome.released += 1, - OutboxPublishFailureAction::Failed => outcome.failed += 1, + Ok(OutboxPublishFailureAction::Released) => { + outcome.released += 1; + context.emit_publish_failure( + service_name, + FailureAction::Release, + "released", + &publish_error, + ); + } + Ok(OutboxPublishFailureAction::Failed) => { + outcome.failed += 1; + context.emit_publish_failure( + service_name, + FailureAction::Fail, + "failed", + &publish_error, + ); + } + Err(error) => { + context.emit_repository_settle_failure( + service_name, + intended_action, + &error, + ); + return Err(error); + } } } } } - store.complete_many(&published).await?; + if let Err(error) = store.complete_many(&published).await { + emit_outbox_repository_failure( + service_name, + FailureOperation::Settle, + FailureAction::SettleAck, + &error, + first_published_context, + ); + return Err(error); + } outcome.published = published.len(); Ok(outcome) } +#[derive(Debug, Clone)] +struct OutboxFailureContext { + message: FailureMessageFields, + attempt: u32, + max_attempts: u32, + source_aggregate_type: String, + source_sequence: String, +} + +impl OutboxFailureContext { + fn from_message(claim: &OutboxClaimRef, max_attempts: u32, message: &Message) -> Self { + Self { + message: FailureMessageFields::from_message(message), + attempt: claim.attempt, + max_attempts, + source_aggregate_type: message + .metadata("x-sourced-source-aggregate-type") + .unwrap_or_default() + .to_string(), + source_sequence: message + .metadata("x-sourced-source-sequence") + .unwrap_or_default() + .to_string(), + } + } + + fn action_for_attempt(&self) -> FailureAction { + if self.attempt >= self.max_attempts { + FailureAction::Fail + } else { + FailureAction::Release + } + } + + fn emit_publish_failure( + &self, + service_name: Option<&str>, + action: FailureAction, + status: &'static str, + error: &TransportError, + ) { + FailureRecord::from_transport_error( + FailureComponent::Outbox, + FailureOperation::Publish, + action, + error, + ) + .with_service(service_name) + .with_message(self.message.clone()) + .with_category(FailureCategory::OutboxPublish) + .with_outbox_fields( + self.attempt, + self.max_attempts, + status, + self.source_aggregate_type.clone(), + self.source_sequence.clone(), + ) + .emit(); + } + + fn emit_repository_settle_failure( + &self, + service_name: Option<&str>, + action: FailureAction, + error: &RepositoryError, + ) { + FailureRecord::from_repository_error( + FailureComponent::Outbox, + FailureOperation::Settle, + action, + error, + ) + .with_service(service_name) + .with_message(self.message.clone()) + .with_category(FailureCategory::OutboxSettle) + .with_outbox_fields( + self.attempt, + self.max_attempts, + action.as_str(), + self.source_aggregate_type.clone(), + self.source_sequence.clone(), + ) + .emit(); + } +} + +fn emit_outbox_repository_failure( + service_name: Option<&str>, + operation: FailureOperation, + action: FailureAction, + error: &RepositoryError, + context: Option, +) { + let mut record = + FailureRecord::from_repository_error(FailureComponent::Outbox, operation, action, error) + .with_service(service_name); + if matches!(operation, FailureOperation::Settle) { + record = record.with_category(FailureCategory::OutboxSettle); + } + if let Some(context) = context { + record = record + .with_message(context.message.clone()) + .with_outbox_fields( + context.attempt, + context.max_attempts, + action.as_str(), + context.source_aggregate_type, + context.source_sequence, + ); + } + record.emit(); +} + /// Publish one mapped outbox message, wrapped in a framework span when the /// `otel` feature is enabled (parented from the message's trace metadata). async fn publish_with_span( @@ -630,6 +825,73 @@ mod tests { assert!(load(&repo, &id).is_failed()); } + #[cfg(feature = "failure-logs")] + #[test] + fn publish_failures_emit_release_and_fail_without_payload_leakage() { + let capture = crate::failure_log::testing::capture_failures(); + + let release_repo = InMemoryRepository::new(); + let mut release = outbox("evt-release-secret"); + release.payload = b"payload-secret-release".to_vec(); + release + .metadata + .insert("authorization".to_string(), "raw-auth".to_string()); + release + .metadata + .insert("arbitrary".to_string(), "metadata-secret".to_string()); + let release_id = store_message(&release_repo, release); + let release_dispatcher = dispatcher(&release_repo, true, 3).with_service("orders"); + + let release_outcome = + block_on(release_dispatcher.dispatch_ids(std::slice::from_ref(&release_id))).unwrap(); + assert_eq!(release_outcome.released, 1); + + let fail_repo = InMemoryRepository::new(); + let mut fail = outbox("evt-fail-secret"); + fail.payload = b"payload-secret-fail".to_vec(); + let fail_id = store_message(&fail_repo, fail); + let fail_dispatcher = dispatcher(&fail_repo, true, 1).with_service("orders"); + + let fail_outcome = + block_on(fail_dispatcher.dispatch_ids(std::slice::from_ref(&fail_id))).unwrap(); + assert_eq!(fail_outcome.failed, 1); + + let events = capture.failure_events(); + let released = events + .iter() + .find(|event| event.field("distributed.failure.action") == Some("release")) + .expect("release failure event"); + assert_eq!(released.field("distributed.component"), Some("outbox")); + assert_eq!(released.field("distributed.operation"), Some("publish")); + assert_eq!(released.field("error.category"), Some("outbox_publish")); + assert_eq!(released.field("outbox.attempt"), Some("1")); + assert_eq!(released.field("outbox.max_attempts"), Some("3")); + assert_eq!(released.field("outbox.status"), Some("released")); + + let failed = events + .iter() + .find(|event| event.field("distributed.failure.action") == Some("fail")) + .expect("fail failure event"); + assert_eq!(failed.field("outbox.attempt"), Some("1")); + assert_eq!(failed.field("outbox.max_attempts"), Some("1")); + assert_eq!(failed.field("outbox.status"), Some("failed")); + + let rendered = format!("{events:?}"); + for forbidden in [ + "payload-secret-release", + "payload-secret-fail", + "raw-auth", + "metadata-secret", + "evt-release-secret", + "evt-fail-secret", + ] { + assert!( + !rendered.contains(forbidden), + "failure event leaked `{forbidden}`: {rendered}" + ); + } + } + #[test] fn dispatch_ids_only_claims_requested_ids() { let repo = InMemoryRepository::new(); diff --git a/src/outbox_worker/publish_hook.rs b/src/outbox_worker/publish_hook.rs index 2250f86..371d5d4 100644 --- a/src/outbox_worker/publish_hook.rs +++ b/src/outbox_worker/publish_hook.rs @@ -65,6 +65,7 @@ where &self.publisher, claimed, self.max_attempts, + self.service_name.as_deref(), std::num::NonZeroUsize::MIN, ) .await?; diff --git a/src/telemetry.rs b/src/telemetry.rs index bff32be..acd87aa 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -122,10 +122,15 @@ pub(crate) mod privacy_policy { "aggregate_id", "aggregate_type", "stream_id", + "raw_path", + "path", "user_id", "tenant_id", "payload", "metadata", + "error", + "error_type", + "error_message", "http_path", "http_target", "http_route",