From 957937ebaa280a68008abe0d6704f031cfdf94f2 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Tue, 7 Jul 2026 02:20:31 -0500 Subject: [PATCH] feat: add private diagnostics snapshot Implements [[tasks/distributed-private-diagnostics-json-1]] --- Cargo.toml | 1 + docs/diagnostics.md | 56 + docs/metrics.md | 4 + docs/observability.md | 7 + src/bus/runner.rs | 77 +- src/diagnostics.rs | 1435 ++++++++++++++++++++++++++ src/lib.rs | 1 + src/microsvc/http.rs | 48 +- src/microsvc/mod.rs | 2 +- src/microsvc/service.rs | 11 + src/outbox_worker/mod.rs | 2 +- src/outbox_worker/outbox_dispatch.rs | 51 +- src/outbox_worker/publish_hook.rs | 1 + tests/diagnostics/main.rs | 306 ++++++ 14 files changed, 1978 insertions(+), 24 deletions(-) create mode 100644 docs/diagnostics.md create mode 100644 src/diagnostics.rs create mode 100644 tests/diagnostics/main.rs diff --git a/Cargo.toml b/Cargo.toml index 7c36648c..0de9c387 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,6 +56,7 @@ serde = { version = "1.0.210", features = ["derive"] } serde_json = "1.0.128" distributed_macros = { workspace = true } sqlx = { version = "0.9", default-features = false, optional = true } +time = { version = "0.3", features = ["formatting"] } tonic = { version = "0.14", default-features = false, features = ["router", "transport", "codegen"], optional = true } tonic-prost = { version = "0.14", optional = true } prost = { version = "0.14", optional = true } diff --git a/docs/diagnostics.md b/docs/diagnostics.md new file mode 100644 index 00000000..b9f30d91 --- /dev/null +++ b/docs/diagnostics.md @@ -0,0 +1,56 @@ +# Diagnostics + +Distributed diagnostics are a private JSON snapshot for operators and agents. +They summarize service inventory, readiness, telemetry capabilities, outbox +backlog, recent sanitized failures, causal hints, and suggested next checks. + +Diagnostics are disabled by default. `microsvc::router(...)` and +`microsvc::serve(...)` do not expose the endpoint. A service must explicitly +compose it: + +```rust +let app = distributed::microsvc::router_with_diagnostics( + service, + distributed::diagnostics::DiagnosticsOptions::new() + .with_bearer_token("management-token"), +); +``` + +The default path is `GET /_distributed/diagnostics`. Responses always include +`Cache-Control: no-store`, even though snapshots are internally cached for a +short TTL. + +## Privacy Boundary + +Treat diagnostics as reconnaissance-grade private data. It includes command and +event names, transport names, readiness state, backlog summaries, recent failure +timing, correlation IDs, trace IDs, and runbook hints. Do not expose it on the +same unauthenticated scrape surface as `/metrics`. + +Use one of these controls: + +- localhost or management-only side port; +- cluster-internal service with NetworkPolicy; +- mTLS or service-mesh authorization; +- trusted proxy authentication; +- the `DiagnosticsOptions::with_access_check(...)` hook or + `with_bearer_token(...)` helper. + +The snapshot intentionally omits payloads, decoded handler input, session +variables, arbitrary metadata, auth headers, cookies, environment variables, DB +URLs, and secrets. Recent failures are sanitized before entering the in-memory +ring buffer and are bounded by capacity, per-response limits, dropped counts, +and truncation metadata. + +## Backlog Provider + +Diagnostics can include outbox backlog state when the service supplies an +outbox store: + +```rust +let diagnostics = distributed::diagnostics::DiagnosticsOptions::new() + .with_outbox_store(repo.outbox_store()); +``` + +The snapshot uses `OutboxStore::backlog_stats`, so it reports count and oldest +age summaries rather than raw rows. diff --git a/docs/metrics.md b/docs/metrics.md index 0c08aac2..762318d1 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -93,6 +93,10 @@ 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. +Private diagnostics are a separate opt-in JSON endpoint, not a Prometheus +scrape route. Do not expose `/_distributed/diagnostics` with the unauthenticated +`/metrics` posture; see [`diagnostics.md`](diagnostics.md). + ## Boundaries The `metrics` feature owns Prometheus text exposition for framework metrics. It diff --git a/docs/observability.md b/docs/observability.md index 8fc2db1a..99a0ed96 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -177,3 +177,10 @@ 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. + +Private diagnostics now live behind explicit composition of +`microsvc::router_with_diagnostics(...)` at `/_distributed/diagnostics`. Treat +that JSON as reconnaissance-grade private data; it must use localhost, +management-only networking, mTLS, trusted-proxy auth, or an access hook and must +not follow `/metrics` unauthenticated exposure semantics. See +[`diagnostics.md`](diagnostics.md). diff --git a/src/bus/runner.rs b/src/bus/runner.rs index c95c62b8..5ad2aafc 100644 --- a/src/bus/runner.rs +++ b/src/bus/runner.rs @@ -69,7 +69,16 @@ where // same as a permanent dispatch failure, so it is dead-lettered/parked. if let Some(error) = received.decode_error() { let action = options.failure_policy.resolve(error); - record_transport_failure(service, transport, error.kind(), action); + let message_context = + crate::diagnostics::FailureMessageContext::from_message(received.message()); + record_transport_failure( + service, + transport, + Some(&message_context), + error.kind(), + action, + Some(error.message()), + ); let kind = received.message().kind; match action { FailureAction::Nack => { @@ -154,7 +163,16 @@ where } Err(error) => match options.failure_policy.resolve(&error) { action @ FailureAction::Nack => { - record_transport_failure(service, transport, error.kind(), action); + let message_context = + crate::diagnostics::FailureMessageContext::from_message(received.message()); + record_transport_failure( + service, + transport, + Some(&message_context), + error.kind(), + action, + Some(error.message()), + ); let reason = error.to_string(); settle_and_record( service, @@ -167,7 +185,16 @@ where .await?; } action @ FailureAction::DeadLetter => { - record_transport_failure(service, transport, error.kind(), action); + let message_context = + crate::diagnostics::FailureMessageContext::from_message(received.message()); + record_transport_failure( + service, + transport, + Some(&message_context), + error.kind(), + action, + Some(error.message()), + ); let reason = error.to_string(); settle_and_record( service, @@ -180,7 +207,16 @@ where .await?; } action @ FailureAction::Park => { - record_transport_failure(service, transport, error.kind(), action); + let message_context = + crate::diagnostics::FailureMessageContext::from_message(received.message()); + record_transport_failure( + service, + transport, + Some(&message_context), + error.kind(), + action, + Some(error.message()), + ); let reason = error.to_string(); settle_and_record( service, @@ -196,8 +232,12 @@ where record_transport_failure( service, transport, + Some(&crate::diagnostics::FailureMessageContext::from_message( + received.message(), + )), error.kind(), FailureAction::LogAndAck, + Some(error.message()), ); eprintln!( "[bus::runner] dropping message '{}' after permanent failure: {error}", @@ -214,7 +254,16 @@ where .await?; } FailureAction::Stop => { - record_transport_failure(service, transport, error.kind(), FailureAction::Stop); + let message_context = + crate::diagnostics::FailureMessageContext::from_message(received.message()); + record_transport_failure( + service, + transport, + Some(&message_context), + error.kind(), + FailureAction::Stop, + Some(error.message()), + ); return Err(error); } }, @@ -244,8 +293,10 @@ where record_transport_failure( service, transport, + None, error.kind(), crate::telemetry::settle_failure_action(settle_action), + Some(error.message()), ); Err(error) } @@ -263,8 +314,10 @@ async fn recv_next( record_transport_failure( service, transport, + None, error.kind(), crate::telemetry::failure_action::RECV_ERROR, + Some(error.message()), ); Err(error) } @@ -329,23 +382,33 @@ fn record_transport_message( fn record_transport_failure( service: Option<&str>, transport: &str, + message: Option<&crate::diagnostics::FailureMessageContext>, kind: TransportErrorKind, action: A, + error_summary: Option<&str>, ) where A: IntoFailureActionLabel, { + let action = action.into_failure_action_label(); #[cfg(feature = "metrics")] crate::metrics::record_transport_failure( service, transport, crate::telemetry::transport_failure_class(kind), - action.into_failure_action_label(), + action, ); #[cfg(not(feature = "metrics"))] { let _ = (service, transport, kind); - let _ = action.into_failure_action_label(); } + crate::diagnostics::record_transport_failure( + service, + transport, + message, + kind, + action, + error_summary, + ); } trait IntoFailureActionLabel { diff --git a/src/diagnostics.rs b/src/diagnostics.rs new file mode 100644 index 00000000..68f8ca86 --- /dev/null +++ b/src/diagnostics.rs @@ -0,0 +1,1435 @@ +//! Private diagnostics snapshots for operators and agents. +//! +//! Diagnostics are not installed into any HTTP router by default. A service must +//! explicitly compose the diagnostics route, and deployments must keep it behind +//! a private listener, trusted proxy, mTLS, or equivalent access control. + +use std::collections::{BTreeSet, VecDeque}; +use std::future::Future; +use std::pin::Pin; +use std::sync::{Arc, Mutex, MutexGuard as StdMutexGuard, OnceLock}; +use std::time::{Duration, SystemTime}; + +#[cfg(feature = "http")] +use axum::http::{header, HeaderMap}; +use serde::{Deserialize, Serialize}; +use time::format_description::well_known::Rfc3339; +use time::OffsetDateTime; + +use crate::bus::{Message, MessageKind, TransportErrorKind}; +use crate::microsvc::{DeliveryKind, HandlerError, HandlerSpec, Service}; +use crate::outbox_worker::{OutboxBacklogStats, OutboxStore, BACKLOG_STATS_SCAN_LIMIT}; +use crate::repository::RepositoryError; +use crate::trace_context::is_valid_traceparent; + +pub const DIAGNOSTICS_SCHEMA_VERSION: u32 = 1; +pub const DEFAULT_DIAGNOSTICS_PATH: &str = "/_distributed/diagnostics"; +const DEFAULT_TTL: Duration = Duration::from_secs(1); +const DEFAULT_RECENT_FAILURE_CAPACITY: usize = 100; +const DEFAULT_RECENT_FAILURE_LIMIT: usize = 32; +const DEFAULT_COMMAND_LIMIT: usize = 64; +const DEFAULT_EVENT_LIMIT: usize = 64; +const DEFAULT_HANDLER_LIMIT: usize = 64; +const DEFAULT_TRANSPORT_LIMIT: usize = 16; +const DEFAULT_METRIC_FAMILY_LIMIT: usize = 64; +const DEFAULT_STRING_LIMIT: usize = 512; +const DEFAULT_ID_LIMIT: usize = 128; +const DEFAULT_RESPONSE_SIZE_LIMIT: usize = 64 * 1024; + +type BacklogStatsFuture = + Pin> + Send>>; +type BacklogStatsProvider = Arc BacklogStatsFuture + Send + Sync>; +#[cfg(feature = "http")] +type AccessCheck = Arc bool + Send + Sync>; + +static RECENT_FAILURES: OnceLock> = OnceLock::new(); + +#[derive(Clone)] +pub struct Diagnostics { + inner: Arc, +} + +struct DiagnosticsInner { + options: DiagnosticsOptions, + cache: Mutex>, +} + +#[derive(Clone)] +struct CachedSnapshot { + generated_at: SystemTime, + snapshot: DiagnosticsSnapshot, +} + +#[derive(Clone)] +pub struct DiagnosticsOptions { + ttl: Duration, + command_limit: usize, + event_limit: usize, + handler_limit: usize, + transport_limit: usize, + metric_family_limit: usize, + recent_failure_limit: usize, + string_limit: usize, + id_limit: usize, + max_response_bytes: usize, + instance_id: Option, + transports: Vec, + diagnostics_path: String, + backlog_stats: Option, + #[cfg(feature = "http")] + access_check: Option, +} + +impl Default for DiagnosticsOptions { + fn default() -> Self { + Self { + ttl: DEFAULT_TTL, + command_limit: DEFAULT_COMMAND_LIMIT, + event_limit: DEFAULT_EVENT_LIMIT, + handler_limit: DEFAULT_HANDLER_LIMIT, + transport_limit: DEFAULT_TRANSPORT_LIMIT, + metric_family_limit: DEFAULT_METRIC_FAMILY_LIMIT, + recent_failure_limit: DEFAULT_RECENT_FAILURE_LIMIT, + string_limit: DEFAULT_STRING_LIMIT, + id_limit: DEFAULT_ID_LIMIT, + max_response_bytes: DEFAULT_RESPONSE_SIZE_LIMIT, + instance_id: None, + transports: Vec::new(), + diagnostics_path: DEFAULT_DIAGNOSTICS_PATH.to_string(), + backlog_stats: None, + #[cfg(feature = "http")] + access_check: None, + } + } +} + +impl DiagnosticsOptions { + pub fn new() -> Self { + Self::default() + } + + pub fn with_ttl(mut self, ttl: Duration) -> Self { + self.ttl = ttl; + self + } + + pub fn with_command_limit(mut self, limit: usize) -> Self { + self.command_limit = limit; + self + } + + pub fn with_event_limit(mut self, limit: usize) -> Self { + self.event_limit = limit; + self + } + + pub fn with_recent_failure_limit(mut self, limit: usize) -> Self { + self.recent_failure_limit = limit; + self + } + + pub fn with_string_limit(mut self, limit: usize) -> Self { + self.string_limit = limit; + self + } + + pub fn with_max_response_bytes(mut self, limit: usize) -> Self { + self.max_response_bytes = limit; + self + } + + pub fn with_instance_id(mut self, instance_id: impl Into) -> Self { + self.instance_id = Some(instance_id.into()); + self + } + + pub fn with_transport(mut self, transport: impl Into) -> Self { + self.transports.push(transport.into()); + self + } + + pub fn with_diagnostics_path(mut self, path: impl Into) -> Self { + self.diagnostics_path = path.into(); + self + } + + #[cfg(feature = "http")] + pub(crate) fn diagnostics_path(&self) -> &str { + &self.diagnostics_path + } + + pub fn with_outbox_store(self, store: S) -> Self + where + S: OutboxStore + 'static, + { + let store = Arc::new(store); + self.with_outbox_backlog_provider(move || { + let store = Arc::clone(&store); + async move { store.backlog_stats().await } + }) + } + + pub fn with_outbox_backlog_provider(mut self, provider: F) -> Self + where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + { + self.backlog_stats = Some(Arc::new(move || Box::pin(provider()))); + self + } + + #[cfg(feature = "http")] + pub fn with_access_check(mut self, access_check: F) -> Self + where + F: Fn(&HeaderMap) -> bool + Send + Sync + 'static, + { + self.access_check = Some(Arc::new(access_check)); + self + } + + #[cfg(feature = "http")] + pub fn with_bearer_token(self, token: impl Into) -> Self { + let expected = format!("Bearer {}", token.into()); + self.with_access_check(move |headers| { + headers + .get(header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value == expected) + }) + } +} + +impl Diagnostics { + pub fn new(options: DiagnosticsOptions) -> Self { + Self { + inner: Arc::new(DiagnosticsInner { + options, + cache: Mutex::new(None), + }), + } + } + + pub async fn snapshot(&self, service: &Service) -> DiagnosticsSnapshot { + let now = SystemTime::now(); + if let Some(snapshot) = self.cached_snapshot(now) { + return snapshot; + } + + let snapshot = build_snapshot(service, &self.inner.options, now).await; + *self.lock_cache() = Some(CachedSnapshot { + generated_at: now, + snapshot: snapshot.clone(), + }); + snapshot + } + + pub fn invalidate(&self) { + *self.lock_cache() = None; + } + + #[cfg(feature = "http")] + pub(crate) fn authorized(&self, headers: &HeaderMap) -> bool { + self.inner + .options + .access_check + .as_ref() + .map(|check| check(headers)) + .unwrap_or(true) + } + + fn cached_snapshot(&self, now: SystemTime) -> Option { + let cached = self.lock_cache().clone()?; + let age = now.duration_since(cached.generated_at).unwrap_or_default(); + if age > self.inner.options.ttl { + return None; + } + Some(with_age(cached.snapshot, cached.generated_at, now)) + } + + fn lock_cache(&self) -> StdMutexGuard<'_, Option> { + self.inner + .cache + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +impl Default for Diagnostics { + fn default() -> Self { + Self::new(DiagnosticsOptions::default()) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct DiagnosticsSnapshot { + pub schema_version: u32, + pub generated_at: String, + pub snapshot: SnapshotMetadata, + pub service: ServiceDiagnostics, + pub health: HealthDiagnostics, + pub telemetry: TelemetryDiagnostics, + pub backlogs: BacklogDiagnostics, + pub recent_failures: Vec, + pub recent_failures_meta: RecentFailuresMetadata, + pub causal_hints: CausalHints, + pub actions: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SnapshotMetadata { + pub age_ms: u64, + pub ttl_ms: u64, + pub partial: bool, + pub truncated: bool, + pub partial_errors: Vec, + pub limits: SnapshotLimits, + pub response_size_bytes: usize, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SnapshotLimits { + pub max_response_bytes: usize, + pub string_bytes: usize, + pub id_bytes: usize, + pub commands: usize, + pub events: usize, + pub handlers: usize, + pub transports: usize, + pub recent_failures: usize, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PartialError { + pub section: String, + pub error_summary: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ServiceDiagnostics { + pub name: String, + pub version: String, + pub instance_id: Option, + pub commands: Vec, + pub commands_meta: ListMetadata, + pub events: Vec, + pub events_meta: ListMetadata, + pub handler_specs: Vec, + pub handler_specs_meta: ListMetadata, + pub subscription_plan: SubscriptionPlanDiagnostics, + pub transports: Vec, + pub transports_meta: ListMetadata, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct HandlerSpecDiagnostic { + pub names: Vec, + pub names_meta: ListMetadata, + pub kind: String, + pub delivery: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SubscriptionPlanDiagnostics { + pub commands: Vec, + pub commands_meta: ListMetadata, + pub events: Vec, + pub events_meta: ListMetadata, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ListMetadata { + pub limit: usize, + pub omitted_count: usize, + pub truncated: bool, +} + +impl ListMetadata { + fn new(limit: usize, total: usize, returned: usize) -> Self { + let omitted_count = total.saturating_sub(returned); + Self { + limit, + omitted_count, + truncated: omitted_count > 0, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct HealthDiagnostics { + pub status: String, + pub readiness: ReadinessDiagnostics, + pub checks: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ReadinessDiagnostics { + pub accepts_commands: bool, + pub consumes_bus: bool, + pub publishes_outbox: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct HealthCheck { + pub id: String, + pub status: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub summary: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct TelemetryDiagnostics { + pub metrics: MetricsTelemetry, + pub tracing: TracingTelemetry, + pub diagnostics: DiagnosticsTelemetry, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct MetricsTelemetry { + pub enabled: bool, + pub path: Option, + pub families: Vec, + pub families_meta: ListMetadata, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct TracingTelemetry { + pub propagation: String, + pub otel_spans: bool, + pub export: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct DiagnosticsTelemetry { + pub enabled: bool, + pub path: String, + pub visibility: String, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct BacklogDiagnostics { + pub outbox: OutboxBacklogDiagnostics, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct OutboxBacklogDiagnostics { + pub configured: bool, + pub pending: Option, + pub oldest_pending_age_seconds: Option, + pub scan_limit: usize, + pub truncated: bool, + pub error: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RecentFailure { + pub at: String, + pub service: String, + pub kind: String, + pub component: String, + pub operation: String, + pub class: String, + pub action: String, + pub transport: Option, + pub message_kind: Option, + pub message: String, + pub error_category: String, + pub error_summary: String, + pub correlation_id: Option, + pub causation_id: Option, + pub trace_id: Option, + pub parent_span_id: Option, + pub trace_flags: Option, + pub runbook_keys: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RecentFailuresMetadata { + pub capacity: usize, + pub dropped_count: u64, + pub limit: usize, + pub omitted_count: usize, + pub truncated: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct CausalHints { + pub last_trace_ids: Vec, + pub last_correlation_ids: Vec, + pub notes: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SuggestedAction { + pub id: String, + pub severity: String, + pub evidence: Vec, + pub suggested_next_steps: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FailureMessageContext { + name: String, + kind: MessageKind, + correlation_id: Option, + causation_id: Option, + traceparent: Option, +} + +impl FailureMessageContext { + pub(crate) fn new(name: impl Into, kind: MessageKind) -> Self { + Self { + name: name.into(), + kind, + correlation_id: None, + causation_id: None, + traceparent: None, + } + } + + pub(crate) fn from_message(message: &Message) -> Self { + Self { + name: message.name().to_string(), + kind: message.kind, + correlation_id: message.correlation_id().map(str::to_string), + causation_id: message.causation_id().map(str::to_string), + traceparent: message.traceparent().map(str::to_string), + } + } +} + +#[derive(Clone, Debug)] +pub struct RecentFailureRingBuffer { + capacity: usize, + records: VecDeque, + dropped_count: u64, +} + +impl RecentFailureRingBuffer { + pub fn new(capacity: usize) -> Self { + Self { + capacity, + records: VecDeque::with_capacity(capacity), + dropped_count: 0, + } + } + + pub fn push(&mut self, failure: RecentFailure) { + if self.capacity == 0 { + self.dropped_count = self.dropped_count.saturating_add(1); + return; + } + if self.records.len() >= self.capacity { + self.records.pop_front(); + self.dropped_count = self.dropped_count.saturating_add(1); + } + self.records.push_back(failure); + } + + pub fn snapshot_for_service( + &self, + service: &str, + limit: usize, + ) -> (Vec, RecentFailuresMetadata) { + let matching = self + .records + .iter() + .rev() + .filter(|failure| failure.service == service) + .cloned() + .collect::>(); + let mut returned = matching.into_iter().take(limit).collect::>(); + returned.reverse(); + let total_for_service = self + .records + .iter() + .filter(|failure| failure.service == service) + .count(); + let omitted_count = total_for_service.saturating_sub(returned.len()); + ( + returned, + RecentFailuresMetadata { + capacity: self.capacity, + dropped_count: self.dropped_count, + limit, + omitted_count, + truncated: omitted_count > 0, + }, + ) + } +} + +impl Default for RecentFailureRingBuffer { + fn default() -> Self { + Self::new(DEFAULT_RECENT_FAILURE_CAPACITY) + } +} + +pub(crate) fn record_microsvc_failure( + service: Option<&str>, + message: &FailureMessageContext, + error: &HandlerError, +) { + let action = format!("return_{}", error.status_code()); + let class = transport_error_class(error.transport_error_kind()); + let category = handler_error_category(error); + push_recent_failure(build_failure_record(FailureRecordInput { + service, + kind: "microsvc", + component: "microsvc", + operation: "dispatch", + class, + action: &action, + transport: None, + message: Some(message), + error_category: category, + error_summary: &error.to_string(), + })); +} + +pub(crate) fn record_transport_failure( + service: Option<&str>, + transport: &str, + message: Option<&FailureMessageContext>, + kind: TransportErrorKind, + action: &str, + error_summary: Option<&str>, +) { + push_recent_failure(build_failure_record(FailureRecordInput { + service, + kind: "transport", + component: "transport", + operation: "receive", + class: transport_error_class(kind), + action, + transport: Some(transport), + message, + error_category: "transport", + error_summary: error_summary.unwrap_or(action), + })); +} + +pub(crate) fn record_outbox_publish_failure( + service: Option<&str>, + message: &FailureMessageContext, + kind: TransportErrorKind, + action: &str, + error_summary: &str, +) { + push_recent_failure(build_failure_record(FailureRecordInput { + service, + kind: "outbox", + component: "outbox", + operation: "publish", + class: transport_error_class(kind), + action, + transport: Some("outbox"), + message: Some(message), + error_category: "outbox_publish", + error_summary, + })); +} + +struct FailureRecordInput<'a> { + service: Option<&'a str>, + kind: &'static str, + component: &'static str, + operation: &'static str, + class: &'static str, + action: &'a str, + transport: Option<&'a str>, + message: Option<&'a FailureMessageContext>, + error_category: &'static str, + error_summary: &'a str, +} + +fn build_failure_record(input: FailureRecordInput<'_>) -> RecentFailure { + let message = input + .message + .map(|message| sanitize_label(&message.name, DEFAULT_ID_LIMIT)) + .unwrap_or_else(|| "unknown".to_string()); + let message_kind = input + .message + .map(|message| message.kind.as_str().to_string()); + let traceparent = input + .message + .and_then(|message| message.traceparent.as_deref()); + let trace = traceparent.and_then(parse_traceparent); + let category = input.error_category.to_string(); + RecentFailure { + at: format_timestamp(SystemTime::now()), + service: service_label(input.service), + kind: input.kind.to_string(), + component: input.component.to_string(), + operation: input.operation.to_string(), + class: input.class.to_string(), + action: sanitize_label(input.action, DEFAULT_ID_LIMIT), + transport: input + .transport + .map(|transport| sanitize_label(transport, DEFAULT_ID_LIMIT)), + message_kind, + message, + error_category: category.clone(), + error_summary: sanitize_summary(input.error_summary, DEFAULT_STRING_LIMIT), + correlation_id: input + .message + .and_then(|message| message.correlation_id.as_deref()) + .map(|id| sanitize_id(id, DEFAULT_ID_LIMIT)), + causation_id: input + .message + .and_then(|message| message.causation_id.as_deref()) + .map(|id| sanitize_id(id, DEFAULT_ID_LIMIT)), + trace_id: trace.as_ref().map(|trace| trace.trace_id.clone()), + parent_span_id: trace.as_ref().map(|trace| trace.parent_span_id.clone()), + trace_flags: trace.as_ref().map(|trace| trace.flags.clone()), + runbook_keys: runbook_keys_for(&category, input.action, input.kind), + } +} + +fn push_recent_failure(failure: RecentFailure) { + recent_failures() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .push(failure); +} + +async fn build_snapshot( + service: &Service, + options: &DiagnosticsOptions, + generated_at: SystemTime, +) -> DiagnosticsSnapshot { + let service_name = service_label(service.name()); + let service_inventory = service_diagnostics(service, options); + let (backlogs, partial_errors) = backlog_diagnostics(options).await; + let (recent_failures, recent_failures_meta) = + recent_failure_snapshot(&service_name, options.recent_failure_limit); + let causal_hints = causal_hints(&recent_failures); + let telemetry = telemetry_diagnostics(options); + let health = health_diagnostics(&service_inventory, &backlogs, &partial_errors); + let actions = suggested_actions(&backlogs, &recent_failures); + let truncated = service_inventory.commands_meta.truncated + || service_inventory.events_meta.truncated + || service_inventory.handler_specs_meta.truncated + || service_inventory.transports_meta.truncated + || telemetry.metrics.families_meta.truncated + || recent_failures_meta.truncated + || backlogs.outbox.truncated; + let mut snapshot = DiagnosticsSnapshot { + schema_version: DIAGNOSTICS_SCHEMA_VERSION, + generated_at: format_timestamp(generated_at), + snapshot: SnapshotMetadata { + age_ms: 0, + ttl_ms: duration_ms(options.ttl), + partial: !partial_errors.is_empty(), + truncated, + partial_errors, + limits: SnapshotLimits { + max_response_bytes: options.max_response_bytes, + string_bytes: options.string_limit, + id_bytes: options.id_limit, + commands: options.command_limit, + events: options.event_limit, + handlers: options.handler_limit, + transports: options.transport_limit, + recent_failures: options.recent_failure_limit, + }, + response_size_bytes: 0, + }, + service: service_inventory, + health, + telemetry, + backlogs, + recent_failures, + recent_failures_meta, + causal_hints, + actions, + }; + snapshot.snapshot.response_size_bytes = serialized_size(&snapshot); + if snapshot.snapshot.response_size_bytes > options.max_response_bytes { + snapshot.snapshot.truncated = true; + } + snapshot +} + +fn service_diagnostics(service: &Service, options: &DiagnosticsOptions) -> ServiceDiagnostics { + let (commands, commands_meta) = capped_sorted_strings( + service.command_names().into_iter().map(str::to_string), + options.command_limit, + options.id_limit, + ); + let (events, events_meta) = capped_sorted_strings( + service.event_names().into_iter().map(str::to_string), + options.event_limit, + options.id_limit, + ); + let handler_specs = service + .handler_specs() + .iter() + .take(options.handler_limit) + .map(|spec| handler_spec_diagnostics(spec, options)) + .collect::>(); + let handler_specs_meta = ListMetadata::new( + options.handler_limit, + service.handler_specs().len(), + handler_specs.len(), + ); + let plan = service.subscription_plan(); + let (plan_commands, plan_commands_meta) = + capped_sorted_strings(plan.commands, options.command_limit, options.id_limit); + let (plan_events, plan_events_meta) = + capped_sorted_strings(plan.events, options.event_limit, options.id_limit); + let (transports, transports_meta) = capped_sorted_strings( + options.transports.clone(), + options.transport_limit, + options.id_limit, + ); + + ServiceDiagnostics { + name: service_label(service.name()), + version: env!("CARGO_PKG_VERSION").to_string(), + instance_id: options + .instance_id + .as_deref() + .map(|id| sanitize_id(id, options.id_limit)), + commands, + commands_meta, + events, + events_meta, + handler_specs, + handler_specs_meta, + subscription_plan: SubscriptionPlanDiagnostics { + commands: plan_commands, + commands_meta: plan_commands_meta, + events: plan_events, + events_meta: plan_events_meta, + }, + transports, + transports_meta, + } +} + +fn handler_spec_diagnostics( + spec: &HandlerSpec, + options: &DiagnosticsOptions, +) -> HandlerSpecDiagnostic { + let (names, names_meta) = capped_sorted_strings( + spec.names().into_iter().map(str::to_string), + options.command_limit.max(options.event_limit), + options.id_limit, + ); + HandlerSpecDiagnostic { + names, + names_meta, + kind: spec.kind.as_str().to_string(), + delivery: match spec.delivery { + DeliveryKind::PointToPoint => "point_to_point", + DeliveryKind::FanOut => "fan_out", + } + .to_string(), + } +} + +async fn backlog_diagnostics( + options: &DiagnosticsOptions, +) -> (BacklogDiagnostics, Vec) { + let Some(provider) = &options.backlog_stats else { + return ( + BacklogDiagnostics { + outbox: OutboxBacklogDiagnostics { + configured: false, + pending: None, + oldest_pending_age_seconds: None, + scan_limit: BACKLOG_STATS_SCAN_LIMIT, + truncated: false, + error: None, + }, + }, + Vec::new(), + ); + }; + + match provider().await { + Ok(stats) => ( + BacklogDiagnostics { + outbox: OutboxBacklogDiagnostics { + configured: true, + pending: Some(stats.pending), + oldest_pending_age_seconds: stats + .oldest_created_at + .and_then(|created_at| SystemTime::now().duration_since(created_at).ok()) + .map(|duration| duration.as_secs_f64()), + scan_limit: BACKLOG_STATS_SCAN_LIMIT, + truncated: stats.pending >= BACKLOG_STATS_SCAN_LIMIT, + error: None, + }, + }, + Vec::new(), + ), + Err(error) => { + let summary = sanitize_summary(&error.to_string(), options.string_limit); + ( + BacklogDiagnostics { + outbox: OutboxBacklogDiagnostics { + configured: true, + pending: None, + oldest_pending_age_seconds: None, + scan_limit: BACKLOG_STATS_SCAN_LIMIT, + truncated: false, + error: Some(summary.clone()), + }, + }, + vec![PartialError { + section: "backlogs.outbox".to_string(), + error_summary: summary, + }], + ) + } + } +} + +fn health_diagnostics( + service: &ServiceDiagnostics, + backlogs: &BacklogDiagnostics, + partial_errors: &[PartialError], +) -> HealthDiagnostics { + let consumes_bus = !service.subscription_plan.commands.is_empty() + || !service.subscription_plan.events.is_empty(); + let mut checks = vec![HealthCheck { + id: "router".to_string(), + status: "ok".to_string(), + summary: None, + }]; + checks.push(match &backlogs.outbox { + outbox if outbox.error.is_some() => HealthCheck { + id: "outbox_backlog".to_string(), + status: "degraded".to_string(), + summary: outbox.error.clone(), + }, + outbox if outbox.pending.unwrap_or_default() > 0 => HealthCheck { + id: "outbox_backlog".to_string(), + status: "warning".to_string(), + summary: Some("pending outbox messages present".to_string()), + }, + outbox if outbox.configured => HealthCheck { + id: "outbox_backlog".to_string(), + status: "ok".to_string(), + summary: None, + }, + _ => HealthCheck { + id: "outbox_backlog".to_string(), + status: "unknown".to_string(), + summary: Some("no outbox backlog provider configured".to_string()), + }, + }); + + HealthDiagnostics { + status: if partial_errors.is_empty() { + "ready".to_string() + } else { + "partial".to_string() + }, + readiness: ReadinessDiagnostics { + accepts_commands: !service.commands.is_empty(), + consumes_bus, + publishes_outbox: backlogs.outbox.configured, + }, + checks, + } +} + +fn telemetry_diagnostics(options: &DiagnosticsOptions) -> TelemetryDiagnostics { + let (families, families_meta) = metric_families(options); + TelemetryDiagnostics { + metrics: MetricsTelemetry { + enabled: cfg!(feature = "metrics"), + path: cfg!(feature = "metrics").then(|| "/metrics".to_string()), + families, + families_meta, + }, + tracing: TracingTelemetry { + propagation: "w3c_trace_context".to_string(), + otel_spans: cfg!(feature = "otel"), + export: if cfg!(feature = "otel") { + "otlp" + } else { + "disabled" + } + .to_string(), + }, + diagnostics: DiagnosticsTelemetry { + enabled: true, + path: options.diagnostics_path.clone(), + visibility: "private".to_string(), + }, + } +} + +fn metric_families(options: &DiagnosticsOptions) -> (Vec, ListMetadata) { + #[cfg(feature = "metrics")] + { + let snapshot = crate::metrics::snapshot(); + let names = snapshot + .families() + .iter() + .map(|family| family.family.name.to_string()); + capped_strings_preserve_order(names, options.metric_family_limit, options.id_limit) + } + #[cfg(not(feature = "metrics"))] + { + ( + Vec::new(), + ListMetadata { + limit: options.metric_family_limit, + omitted_count: 0, + truncated: false, + }, + ) + } +} + +fn recent_failure_snapshot( + service: &str, + limit: usize, +) -> (Vec, RecentFailuresMetadata) { + recent_failures() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .snapshot_for_service(service, limit) +} + +fn causal_hints(recent_failures: &[RecentFailure]) -> CausalHints { + let mut trace_ids = BTreeSet::new(); + let mut correlation_ids = BTreeSet::new(); + let mut notes = BTreeSet::new(); + for failure in recent_failures { + if let Some(trace_id) = &failure.trace_id { + trace_ids.insert(trace_id.clone()); + notes.insert("traceparent present".to_string()); + } + if let Some(correlation_id) = &failure.correlation_id { + correlation_ids.insert(correlation_id.clone()); + notes.insert("correlation_id present".to_string()); + } + if failure.causation_id.is_some() { + notes.insert("causation_id present".to_string()); + } + } + CausalHints { + last_trace_ids: trace_ids.into_iter().rev().take(8).collect(), + last_correlation_ids: correlation_ids.into_iter().rev().take(8).collect(), + notes: notes.into_iter().collect(), + } +} + +fn suggested_actions( + backlogs: &BacklogDiagnostics, + recent_failures: &[RecentFailure], +) -> Vec { + let mut actions = Vec::new(); + if backlogs.outbox.pending.unwrap_or_default() > 0 { + actions.push(SuggestedAction { + id: "outbox_backlog_present".to_string(), + severity: "warning".to_string(), + evidence: vec!["outbox.pending > 0".to_string()], + suggested_next_steps: vec![ + "check transport failure counts".to_string(), + "inspect broker connectivity".to_string(), + ], + }); + } + if recent_failures + .iter() + .any(|failure| failure.kind == "transport") + { + actions.push(SuggestedAction { + id: "recent_transport_failures".to_string(), + severity: "warning".to_string(), + evidence: vec!["recent_failures.kind contains transport".to_string()], + suggested_next_steps: vec![ + "check broker connectivity".to_string(), + "inspect failure action and retry class".to_string(), + ], + }); + } + if recent_failures + .iter() + .any(|failure| failure.error_category == "routing") + { + actions.push(SuggestedAction { + id: "unknown_message_recent".to_string(), + severity: "info".to_string(), + evidence: vec!["recent_failures.error_category contains routing".to_string()], + suggested_next_steps: vec![ + "compare producer message name to service inventory".to_string(), + "check generated subscription plan".to_string(), + ], + }); + } + actions +} + +fn with_age( + mut snapshot: DiagnosticsSnapshot, + generated_at: SystemTime, + now: SystemTime, +) -> DiagnosticsSnapshot { + snapshot.snapshot.age_ms = duration_ms(now.duration_since(generated_at).unwrap_or_default()); + snapshot.snapshot.response_size_bytes = serialized_size(&snapshot); + snapshot +} + +fn capped_sorted_strings( + values: I, + limit: usize, + string_limit: usize, +) -> (Vec, ListMetadata) +where + I: IntoIterator, +{ + let sorted = values + .into_iter() + .map(|value| sanitize_label(&value, string_limit)) + .collect::>() + .into_iter() + .collect::>(); + let total = sorted.len(); + let items = sorted.into_iter().take(limit).collect::>(); + let meta = ListMetadata::new(limit, total, items.len()); + (items, meta) +} + +#[cfg(feature = "metrics")] +fn capped_strings_preserve_order( + values: I, + limit: usize, + string_limit: usize, +) -> (Vec, ListMetadata) +where + I: IntoIterator, +{ + let all = values + .into_iter() + .map(|value| sanitize_label(&value, string_limit)) + .collect::>(); + let total = all.len(); + let items = all.into_iter().take(limit).collect::>(); + let meta = ListMetadata::new(limit, total, items.len()); + (items, meta) +} + +fn recent_failures() -> &'static Mutex { + RECENT_FAILURES.get_or_init(|| Mutex::new(RecentFailureRingBuffer::default())) +} + +fn handler_error_category(error: &HandlerError) -> &'static str { + match error { + HandlerError::UnknownCommand(_) => "routing", + HandlerError::DecodeFailed(_) => "decode", + HandlerError::Rejected(_) => "validation", + HandlerError::NotFound(_) => "repository", + HandlerError::Unauthorized(_) => "auth", + HandlerError::Repository(_) => "repository", + HandlerError::GuardRejected(_) => "guard", + HandlerError::Other(_) => "handler", + } +} + +fn transport_error_class(kind: TransportErrorKind) -> &'static str { + match kind { + TransportErrorKind::Retryable => "retryable", + TransportErrorKind::Permanent => "permanent", + } +} + +fn runbook_keys_for(category: &str, action: &str, kind: &str) -> Vec { + let mut keys = Vec::new(); + match category { + "routing" => keys.extend(["check_service_inventory", "check_message_name"]), + "decode" => keys.extend(["check_payload_codec", "check_producer_schema"]), + "auth" => keys.extend(["check_auth_proxy"]), + "repository" | "storage" => keys.extend(["check_store_connectivity"]), + "transport" => keys.extend(["check_broker_connectivity"]), + "outbox_publish" => keys.extend(["check_broker_connectivity", "inspect_outbox_backlog"]), + _ => keys.extend(["inspect_recent_failure"]), + } + if action == "nack" || action == "release" { + keys.push("watch_retry_backoff"); + } + if kind == "outbox" { + keys.push("inspect_outbox_backlog"); + } + keys.sort_unstable(); + keys.dedup(); + keys.into_iter().map(str::to_string).collect() +} + +fn sanitize_id(value: &str, limit: usize) -> String { + sanitize_label(value, limit) +} + +fn sanitize_label(value: &str, limit: usize) -> String { + if looks_sensitive(value) { + return "[redacted]".to_string(); + } + truncate_chars(value, limit) +} + +fn sanitize_summary(value: &str, limit: usize) -> String { + if looks_sensitive(value) { + return "[redacted]".to_string(); + } + truncate_chars(value, limit) +} + +fn looks_sensitive(value: &str) -> bool { + let lower = value.to_ascii_lowercase(); + let sensitive_terms = [ + "authorization", + "bearer ", + "cookie", + "password", + "passwd", + "secret", + "token", + "api_key", + "apikey", + "private_key", + "database_url", + "db_url", + "postgres://", + "postgresql://", + "mysql://", + "redis://", + "amqp://", + ]; + sensitive_terms.iter().any(|term| lower.contains(term)) +} + +fn truncate_chars(value: &str, limit: usize) -> String { + let mut out = String::new(); + for ch in value.chars().take(limit) { + if ch.is_control() { + out.push(' '); + } else { + out.push(ch); + } + } + out +} + +fn parse_traceparent(value: &str) -> Option { + if !is_valid_traceparent(value) { + return None; + } + let mut parts = value.split('-'); + let _version = parts.next()?; + Some(TraceParts { + trace_id: parts.next()?.to_string(), + parent_span_id: parts.next()?.to_string(), + flags: parts.next()?.to_string(), + }) +} + +struct TraceParts { + trace_id: String, + parent_span_id: String, + flags: String, +} + +fn service_label(service: Option<&str>) -> String { + service + .filter(|value| !value.is_empty()) + .unwrap_or("unnamed") + .to_string() +} + +fn format_timestamp(time: SystemTime) -> String { + OffsetDateTime::from(time) + .format(&Rfc3339) + .unwrap_or_else(|_| "1970-01-01T00:00:00Z".to_string()) +} + +fn duration_ms(duration: Duration) -> u64 { + duration.as_millis().min(u128::from(u64::MAX)) as u64 +} + +fn serialized_size(snapshot: &DiagnosticsSnapshot) -> usize { + serde_json::to_vec(snapshot).map_or(0, |json| json.len()) +} + +#[cfg(test)] +fn reset_for_tests() { + *recent_failures() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = RecentFailureRingBuffer::default(); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::microsvc::{Context, Routes}; + use serde_json::json; + use std::sync::OnceLock; + + static TEST_LOCK: OnceLock> = OnceLock::new(); + + fn lock_for_tests() -> StdMutexGuard<'static, ()> { + TEST_LOCK + .get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn failure(service: &str, message: &str) -> RecentFailure { + RecentFailure { + at: "2026-07-07T00:00:00Z".to_string(), + service: service.to_string(), + kind: "transport".to_string(), + component: "transport".to_string(), + operation: "receive".to_string(), + class: "retryable".to_string(), + action: "nack".to_string(), + transport: Some("nats".to_string()), + message_kind: Some("event".to_string()), + message: message.to_string(), + error_category: "transport".to_string(), + error_summary: "timed out".to_string(), + correlation_id: None, + causation_id: None, + trace_id: None, + parent_span_id: None, + trace_flags: None, + runbook_keys: vec!["check_broker_connectivity".to_string()], + } + } + + fn service() -> Service { + Service::new().named("diag-unit").routes( + Routes::new() + .with_dependencies(()) + .command("orders.create") + .handle(|_ctx: &Context<()>| async move { Ok(json!({"ok": true})) }), + ) + } + + #[test] + fn recent_failure_ring_buffer_evicts_oldest_and_counts_drops() { + let mut buffer = RecentFailureRingBuffer::new(2); + + buffer.push(failure("orders", "first")); + buffer.push(failure("orders", "second")); + buffer.push(failure("orders", "third")); + + let (records, meta) = buffer.snapshot_for_service("orders", 10); + assert_eq!( + records + .iter() + .map(|record| record.message.as_str()) + .collect::>(), + vec!["second", "third"] + ); + assert_eq!(meta.capacity, 2); + assert_eq!(meta.dropped_count, 1); + } + + #[test] + fn recent_failure_snapshot_reports_omitted_count_when_limited() { + let mut buffer = RecentFailureRingBuffer::new(5); + for message in ["one", "two", "three"] { + buffer.push(failure("orders", message)); + } + + let (records, meta) = buffer.snapshot_for_service("orders", 2); + + assert_eq!( + records + .iter() + .map(|record| record.message.as_str()) + .collect::>(), + vec!["two", "three"] + ); + assert_eq!(meta.omitted_count, 1); + assert!(meta.truncated); + } + + #[test] + fn sanitizer_redacts_obvious_secret_material() { + assert_eq!( + sanitize_summary( + "DATABASE_URL=postgres://user:secret@example/db token=abc", + DEFAULT_STRING_LIMIT + ), + "[redacted]" + ); + assert_eq!( + sanitize_label("orders.created", DEFAULT_ID_LIMIT), + "orders.created" + ); + } + + #[tokio::test] + async fn snapshot_cache_reuses_generated_at_until_ttl_expires() { + let _guard = lock_for_tests(); + reset_for_tests(); + let service = service(); + let diagnostics = + Diagnostics::new(DiagnosticsOptions::new().with_ttl(Duration::from_secs(60))); + + let first = diagnostics.snapshot(&service).await; + tokio::time::sleep(Duration::from_millis(15)).await; + let second = diagnostics.snapshot(&service).await; + + assert_eq!(first.generated_at, second.generated_at); + assert!(second.snapshot.age_ms >= first.snapshot.age_ms); + + diagnostics.invalidate(); + let third = diagnostics.snapshot(&service).await; + assert_ne!(second.snapshot.age_ms, third.snapshot.age_ms); + } + + #[tokio::test] + async fn snapshot_marks_partial_outbox_errors_and_bounds_response() { + let _guard = lock_for_tests(); + reset_for_tests(); + let service = service(); + let diagnostics = Diagnostics::new( + DiagnosticsOptions::new() + .with_recent_failure_limit(2) + .with_max_response_bytes(16 * 1024) + .with_outbox_backlog_provider(|| async { + Err(RepositoryError::Model( + "DATABASE_URL=postgres://secret".to_string(), + )) + }), + ); + + let snapshot = diagnostics.snapshot(&service).await; + let json = serde_json::to_string(&snapshot).unwrap(); + + assert!(snapshot.snapshot.partial); + assert_eq!( + snapshot.snapshot.partial_errors[0].section, + "backlogs.outbox" + ); + assert!(!json.contains("postgres://secret")); + assert!(json.len() <= snapshot.snapshot.limits.max_response_bytes); + } + + #[tokio::test] + async fn command_lists_are_capped_with_omitted_metadata() { + let service = Service::new().named("diag-capped").routes( + Routes::new() + .with_dependencies(()) + .command("orders.a") + .handle(|_ctx: &Context<()>| async move { Ok(json!({})) }) + .command("orders.b") + .handle(|_ctx: &Context<()>| async move { Ok(json!({})) }), + ); + let diagnostics = Diagnostics::new(DiagnosticsOptions::new().with_command_limit(1)); + + let snapshot = diagnostics.snapshot(&service).await; + + assert_eq!(snapshot.service.commands.len(), 1); + assert_eq!(snapshot.service.commands_meta.omitted_count, 1); + assert!(snapshot.snapshot.truncated); + } +} diff --git a/src/lib.rs b/src/lib.rs index d6a0112c..afbdda01 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,6 +6,7 @@ extern crate self as distributed; pub mod aggregate; pub mod bus; +pub mod diagnostics; pub mod entity; pub mod repository; diff --git a/src/microsvc/http.rs b/src/microsvc/http.rs index 68bf25bb..89960e35 100644 --- a/src/microsvc/http.rs +++ b/src/microsvc/http.rs @@ -34,7 +34,7 @@ use std::sync::Arc; use axum::extract::{DefaultBodyLimit, Path, State}; -use axum::http::{HeaderMap, StatusCode}; +use axum::http::{header, HeaderMap, HeaderValue, StatusCode}; use axum::response::IntoResponse; use axum::routing::get; use axum::{Json, Router}; @@ -44,6 +44,7 @@ use super::error::HandlerError; use super::service::Service; use super::session::Session; use super::MAX_HTTP_BODY_BYTES; +use crate::diagnostics::{Diagnostics, DiagnosticsOptions}; /// Build an axum `Router` that dispatches commands via the given service. pub fn router(service: Arc) -> Router { @@ -60,6 +61,26 @@ pub fn router(service: Arc) -> Router { .with_state(service) } +/// Build an axum `Router` with command dispatch plus the private diagnostics +/// endpoint. +/// +/// Diagnostics are disabled in [`router`] and [`serve`]. Use this helper only +/// on private listeners or with a configured access hook; the endpoint exposes +/// service inventory and recent sanitized failures. +pub fn router_with_diagnostics(service: Arc, options: DiagnosticsOptions) -> Router { + let options = options.with_transport("http"); + let path = options.diagnostics_path().to_string(); + let diagnostics = Diagnostics::new(options); + let diagnostics_router = Router::new() + .route(&path, get(diagnostics_handler)) + .with_state(DiagnosticsHttpState { + service: service.clone(), + diagnostics, + }); + + router(service).merge(diagnostics_router) +} + /// Serve the service over HTTP at the given address (e.g. `"0.0.0.0:3000"`). pub async fn serve(service: Arc, addr: &str) -> Result<(), std::io::Error> { let app = router(service); @@ -83,6 +104,31 @@ async fn metrics_handler(State(service): State>) -> impl IntoRespon crate::metrics::prometheus_response(service.name()) } +#[derive(Clone)] +struct DiagnosticsHttpState { + service: Arc, + diagnostics: Diagnostics, +} + +async fn diagnostics_handler( + State(state): State, + headers: HeaderMap, +) -> impl IntoResponse { + let mut response = if state.diagnostics.authorized(&headers) { + Json(state.diagnostics.snapshot(&state.service).await).into_response() + } else { + ( + StatusCode::UNAUTHORIZED, + Json(json!({ "error": "unauthorized" })), + ) + .into_response() + }; + response + .headers_mut() + .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store")); + response +} + /// `POST /{command}` — dispatch a command with JSON body and headers as session. async fn command_handler( State(service): State>, diff --git a/src/microsvc/mod.rs b/src/microsvc/mod.rs index 84841d44..237fe504 100644 --- a/src/microsvc/mod.rs +++ b/src/microsvc/mod.rs @@ -92,7 +92,7 @@ pub const MAX_HTTP_BODY_BYTES: usize = 1024 * 1024; #[cfg(feature = "http")] mod http; #[cfg(feature = "http")] -pub use http::{router, serve}; +pub use http::{router, router_with_diagnostics, serve}; // Knative / CloudEvents HTTP ingress (Service-coupled; the bus keeps only the // produce/manifest helpers). Requires the "http" feature. diff --git a/src/microsvc/service.rs b/src/microsvc/service.rs index 6efea9a3..1de531c5 100644 --- a/src/microsvc/service.rs +++ b/src/microsvc/service.rs @@ -586,6 +586,13 @@ impl Service { started.elapsed(), ); } + if let Err(error) = &result { + let message = crate::diagnostics::FailureMessageContext::new( + command.to_string(), + MessageKind::Command, + ); + crate::diagnostics::record_microsvc_failure(self.name(), &message, error); + } result } @@ -656,6 +663,10 @@ impl Service { started.elapsed(), ); } + if let Err(error) = &result { + let message = crate::diagnostics::FailureMessageContext::from_message(message); + crate::diagnostics::record_microsvc_failure(self.name(), &message, error); + } result } diff --git a/src/outbox_worker/mod.rs b/src/outbox_worker/mod.rs index b00c4c28..fce802b9 100644 --- a/src/outbox_worker/mod.rs +++ b/src/outbox_worker/mod.rs @@ -38,7 +38,7 @@ mod testing; pub(crate) use store::ensure_active_claim; pub use store::{ ClaimOutboxMessages, OutboxBacklogStats, OutboxClaimRef, OutboxPublishFailureAction, - OutboxStore, + OutboxStore, BACKLOG_STATS_SCAN_LIMIT, }; // Outbox -> bus bridge (moved out of the bus module; depends up on bus traits). diff --git a/src/outbox_worker/outbox_dispatch.rs b/src/outbox_worker/outbox_dispatch.rs index 89ee6ec6..dcdc53a0 100644 --- a/src/outbox_worker/outbox_dispatch.rs +++ b/src/outbox_worker/outbox_dispatch.rs @@ -253,6 +253,7 @@ where claimed, self.max_attempts, self.publish_concurrency, + self.service_name.as_deref(), ) .await?; self.record_outbox_outcomes(&settled); @@ -355,6 +356,7 @@ pub(crate) async fn publish_and_settle( claimed: Vec, max_attempts: u32, publish_concurrency: NonZeroUsize, + service_name: Option<&str>, ) -> Result where S: OutboxStore, @@ -363,36 +365,57 @@ 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 diagnostics = crate::diagnostics::FailureMessageContext::from_message(&message); + work.push((claim, message, diagnostics)); } // 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, + crate::diagnostics::FailureMessageContext, + Result<(), TransportError>, + )> = stream::iter( + work.into_iter() + .map(|(claim, message, diagnostics)| async move { + let result = publish_with_span(publisher, message).await; + (claim, diagnostics, 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 { + for (claim, diagnostics, result) in results { match result { Ok(()) => published.push(claim), Err(publish_error) => { - match store + let action = match store .record_failure(&claim, &publish_error.to_string(), max_attempts) .await? { - OutboxPublishFailureAction::Released => outcome.released += 1, - OutboxPublishFailureAction::Failed => outcome.failed += 1, - } + OutboxPublishFailureAction::Released => { + outcome.released += 1; + "release" + } + OutboxPublishFailureAction::Failed => { + outcome.failed += 1; + "fail" + } + }; + crate::diagnostics::record_outbox_publish_failure( + service_name, + &diagnostics, + publish_error.kind(), + action, + publish_error.message(), + ); } } } diff --git a/src/outbox_worker/publish_hook.rs b/src/outbox_worker/publish_hook.rs index 2250f86b..110fe2cb 100644 --- a/src/outbox_worker/publish_hook.rs +++ b/src/outbox_worker/publish_hook.rs @@ -66,6 +66,7 @@ where claimed, self.max_attempts, std::num::NonZeroUsize::MIN, + self.service_name.as_deref(), ) .await?; self.record_outbox_outcomes(&settled).await; diff --git a/tests/diagnostics/main.rs b/tests/diagnostics/main.rs new file mode 100644 index 00000000..c238e488 --- /dev/null +++ b/tests/diagnostics/main.rs @@ -0,0 +1,306 @@ +#![cfg(all(feature = "http", feature = "metrics"))] + +use std::collections::VecDeque; +use std::sync::Arc; +use std::time::Duration; + +use distributed::bus::{ + run_source, Handlers, MessagePublisher, MessageSource, ReceivedMessage, RunOptions, + TransportError, +}; +use distributed::diagnostics::{Diagnostics, DiagnosticsOptions, DEFAULT_DIAGNOSTICS_PATH}; +use distributed::microsvc::{self, Context, Message, MessageKind, Routes, Service, Session}; +use distributed::{ + CommitBatch, InMemoryRepository, OutboxDispatcher, OutboxMessage, TransactionalCommit, +}; +use serde_json::json; + +const TRACEPARENT: &str = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + +fn orders_service(name: &str) -> Arc { + Arc::new( + Service::new().named(name).routes( + Routes::new() + .with_dependencies(()) + .command("orders.create") + .handle(|_ctx: &Context<()>| async move { Ok(json!({ "ok": true })) }), + ), + ) +} + +async fn start_app(app: axum::Router) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://{addr}") +} + +#[tokio::test] +async fn default_router_does_not_expose_diagnostics() { + let base = start_app(microsvc::router(orders_service("diag-disabled"))).await; + let response = reqwest::Client::new() + .get(format!("{base}{DEFAULT_DIAGNOSTICS_PATH}")) + .send() + .await + .unwrap(); + + assert_eq!(response.status(), 404); +} + +#[tokio::test] +async fn diagnostics_route_requires_configured_access_hook_and_disables_caching() { + let service = orders_service("diag-http-auth"); + let base = start_app(microsvc::router_with_diagnostics( + service, + DiagnosticsOptions::new().with_bearer_token("correct-token"), + )) + .await; + let client = reqwest::Client::new(); + + let unauthorized = client + .get(format!("{base}{DEFAULT_DIAGNOSTICS_PATH}")) + .send() + .await + .unwrap(); + assert_eq!(unauthorized.status(), 401); + assert_eq!( + unauthorized + .headers() + .get("cache-control") + .and_then(|value| value.to_str().ok()), + Some("no-store") + ); + + let authorized = client + .get(format!("{base}{DEFAULT_DIAGNOSTICS_PATH}")) + .bearer_auth("correct-token") + .send() + .await + .unwrap(); + assert_eq!(authorized.status(), 200); + assert_eq!( + authorized + .headers() + .get("cache-control") + .and_then(|value| value.to_str().ok()), + Some("no-store") + ); + let body: serde_json::Value = authorized.json().await.unwrap(); + assert_eq!(body["schema_version"], 1); + assert_eq!(body["telemetry"]["diagnostics"]["visibility"], "private"); +} + +struct SecretFailPublisher; + +impl MessagePublisher for SecretFailPublisher { + async fn publish(&self, message: Message) -> Result<(), TransportError> { + if message.id().is_some_and(|id| id.contains("poison")) { + return Err(TransportError::retryable( + "DATABASE_URL=postgres://user:pass@localhost/db token=super-secret", + )); + } + Ok(()) + } +} + +struct OneShotSource { + queue: VecDeque, +} + +impl OneShotSource { + fn new(message: Message) -> Self { + Self { + queue: VecDeque::from([message]), + } + } +} + +impl MessageSource for OneShotSource { + type Received = OneShotReceived; + + fn transport_name(&self) -> &'static str { + "nats" + } + + async fn recv(&mut self) -> Result, TransportError> { + Ok(self + .queue + .pop_front() + .map(|message| OneShotReceived { message })) + } +} + +struct OneShotReceived { + message: Message, +} + +impl ReceivedMessage for OneShotReceived { + fn message(&self) -> &Message { + &self.message + } + + async fn ack(self) -> Result<(), TransportError> { + Ok(()) + } + + async fn nack(self, _reason: &str) -> Result<(), TransportError> { + Ok(()) + } +} + +#[tokio::test] +async fn diagnostics_snapshot_agrees_with_metrics_backlog_and_redacts_private_data() { + let service_name = "diag-integration"; + let service = orders_service(service_name); + let repo = InMemoryRepository::new(); + std::env::set_var("DISTRIBUTED_DIAGNOSTICS_SECRET", "env-secret"); + + service + .dispatch( + "orders.create", + json!({ "id": "ok", "decoded": "decoded-input-secret" }), + Session::new(), + ) + .await + .unwrap(); + let mut session = std::collections::HashMap::new(); + session.insert("x-hasura-user-id".to_string(), "session-secret".to_string()); + session.insert( + "authorization".to_string(), + "Bearer auth-secret".to_string(), + ); + let _ = service + .dispatch( + "orders.missing", + json!({ "payload": "payload-secret" }), + Session::from_map(session), + ) + .await; + + let mut batch = CommitBatch::empty(); + for id in ["evt-ok", "evt-poison"] { + batch + .outbox_messages + .push(OutboxMessage::create(id, "orders.created", b"payload-secret".to_vec()).unwrap()); + } + repo.commit_batch(batch).await.unwrap(); + let dispatcher = OutboxDispatcher::new( + repo.outbox_store(), + SecretFailPublisher, + "diag-worker", + Duration::from_secs(60), + 3, + ) + .with_service(service_name); + let outcome = dispatcher.dispatch_batch(10).await.unwrap(); + assert_eq!(outcome.published, 1); + assert_eq!(outcome.released, 1); + + let transport_message = Message::new( + "orders.created", + MessageKind::Event, + br#"{ "payload": "payload-secret" }"#.to_vec(), + ) + .with_metadata("correlation_id", "corr-123") + .with_metadata("causation_id", "cause-456") + .with_metadata("traceparent", TRACEPARENT) + .with_metadata("authorization", "Bearer auth-secret") + .with_metadata("cookie", "cookie-secret") + .with_metadata("x-raw", "raw-meta-secret"); + let handlers = Arc::new(Handlers::new().named(service_name).on_event( + "orders.created", + |_message: &Message| async move { + Err(TransportError::retryable( + "nats timeout token=transport-secret", + )) + }, + )); + run_source( + handlers, + OneShotSource::new(transport_message), + RunOptions::idempotent(), + ) + .await + .unwrap(); + + let diagnostics = Diagnostics::new( + DiagnosticsOptions::new() + .with_outbox_store(repo.outbox_store()) + .with_transport("http") + .with_transport("nats"), + ); + let snapshot = diagnostics.snapshot(&service).await; + + assert_eq!(snapshot.schema_version, 1); + assert!(snapshot + .service + .commands + .contains(&"orders.create".to_string())); + assert!(snapshot.telemetry.metrics.enabled); + assert!(snapshot + .telemetry + .metrics + .families + .contains(&"distributed_microsvc_dispatch_total".to_string())); + assert_eq!(snapshot.backlogs.outbox.pending, Some(1)); + assert!(snapshot + .recent_failures + .iter() + .any(|failure| failure.kind == "microsvc")); + assert!(snapshot + .recent_failures + .iter() + .any(|failure| failure.kind == "transport")); + assert!(snapshot + .recent_failures + .iter() + .any(|failure| failure.kind == "outbox")); + assert!(snapshot + .causal_hints + .last_trace_ids + .contains(&"4bf92f3577b34da6a3ce929d0e0e4736".to_string())); + assert!(snapshot + .causal_hints + .last_correlation_ids + .contains(&"corr-123".to_string())); + assert!(snapshot + .actions + .iter() + .any(|action| action.id == "outbox_backlog_present")); + + let metrics = distributed::metrics::prometheus_text(); + assert!( + metrics.contains("distributed_outbox_pending_messages{service=\"diag-integration\"} 1"), + "metrics should agree with diagnostics backlog:\n{metrics}" + ); + assert!( + metrics.contains("message=\"unknown\""), + "unknown command metric should use the bounded label:\n{metrics}" + ); + + let json = serde_json::to_string(&snapshot).unwrap(); + for forbidden in [ + "payload-secret", + "decoded-input-secret", + "session-secret", + "auth-secret", + "cookie-secret", + "raw-meta-secret", + "transport-secret", + "super-secret", + "postgres://user:pass", + "DATABASE_URL", + "env-secret", + "authorization", + "cookie", + "x-hasura", + ] { + assert!( + !json.contains(forbidden), + "diagnostics leaked forbidden value `{forbidden}`:\n{json}" + ); + } + assert!(json.len() <= snapshot.snapshot.limits.max_response_bytes); +}