diff --git a/TESTING.md b/TESTING.md index 0e4aee87841..d939265c414 100644 --- a/TESTING.md +++ b/TESTING.md @@ -358,7 +358,7 @@ CLI-side, only two matter for testing: | Symptom | Cause | Fix | |---------|-------|-----| | `relay error 500` or `400: restricted: not a channel member` after a code change | Stale binary | Rebuild and re-export `PATH`; or `cargo run` directly | -| `Address already in use` on relay start (os error 48 on macOS, 98 on Linux) | Another relay (or stale process) holding `:3000` / `:8080` / `:9102` (or your override ports) | The panic line names the failing port — read it first. Then `lsof -iTCP:3000,8080,9102 -sTCP:LISTEN` (or your override equivalents). Kill the offender (`pkill -f buzz-relay`) or use the port-override block in step 3. If you already overrode and *still* collide, a prior reviewer left a relay running on the same alt ports — kill it or pick fresh ports | +| `Address already in use` on relay start (os error 48 on macOS, 98 on Linux) | Another relay (or stale process) holding `:3000` / `:8080` / `:9102` (or your override ports) | Metrics-listener failures emit a `metrics_bind` lifecycle terminal with reason `bind`. Check the configured ports with `lsof -iTCP:3000,8080,9102 -sTCP:LISTEN` (or your override equivalents). Kill the offender (`pkill -f buzz-relay`) or use the port-override block in step 3. If you already overrode and *still* collide, a prior reviewer left a relay running on the same alt ports — kill it or pick fresh ports | | `auth_error: BUZZ_PRIVATE_KEY is required` | Env not exported into the CLI's shell | `export BUZZ_PRIVATE_KEY=...` (or pass `--private-key`) | | `auth_error: BUZZ_AUTH_TAG verification failed … signature verification failed` | A stale `BUZZ_AUTH_TAG` inherited from a parent shell. The local dev relay rejects it. | `unset BUZZ_AUTH_TAG` (see the scrub block in step 1) | | `auth-required: verification failed` on a closed relay | NIP-OA attestation needed | Set `BUZZ_AUTH_TAG` to the owner-issued JSON, or relax `BUZZ_REQUIRE_RELAY_MEMBERSHIP` | diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 123440c0416..18ea187fc7d 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -25,6 +25,8 @@ pub mod error; pub mod handlers; /// Stateless HMAC-signed relay invite tokens (mint/verify). pub mod invite_token; +/// Fixed-schema evidence for the relay's earliest startup steps. +pub mod lifecycle; /// Inter-relay mesh startup wiring (`BUZZ_MESH` seam). pub mod mesh_boot; /// Prometheus metrics: recorder, upkeep, HTTP middleware. diff --git a/crates/buzz-relay/src/lifecycle.rs b/crates/buzz-relay/src/lifecycle.rs new file mode 100644 index 00000000000..bc9d51062b2 --- /dev/null +++ b/crates/buzz-relay/src/lifecycle.rs @@ -0,0 +1,591 @@ +//! Fixed-schema evidence for the relay's earliest startup steps. +//! +//! These events are written directly to stderr because crypto, tracing, +//! configuration, and metrics setup can fail before the normal telemetry +//! stack exists. Values are closed enums; raw errors and secrets never enter +//! the lifecycle schema. + +use std::{ + io::Write as _, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + }, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; + +use serde::Serialize; +use uuid::Uuid; + +const EVENT_NAME: &str = "buzz_process_lifecycle"; +const SCHEMA_VERSION: u8 = 1; + +/// A bounded early-startup phase. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum StartupPhase { + /// Process entry through a usable metrics listener. + ProcessTelemetry, + /// Install the process-wide rustls provider. + CryptoInit, + /// Install structured logging and optional OTLP tracing. + TracingInit, + /// Parse environment-backed configuration. + ConfigLoad, + /// Load and validate relay key material. + KeyLoad, + /// Install the Prometheus recorder and bind its listener. + MetricsBind, +} + +impl StartupPhase { + /// The complete wire vocabulary. + pub const ALL: [Self; 6] = [ + Self::ProcessTelemetry, + Self::CryptoInit, + Self::TracingInit, + Self::ConfigLoad, + Self::KeyLoad, + Self::MetricsBind, + ]; + + /// Stable wire value. + pub const fn as_str(self) -> &'static str { + match self { + Self::ProcessTelemetry => "process_telemetry", + Self::CryptoInit => "crypto_init", + Self::TracingInit => "tracing_init", + Self::ConfigLoad => "config_load", + Self::KeyLoad => "key_load", + Self::MetricsBind => "metrics_bind", + } + } +} + +/// A bounded terminal status. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum LifecycleStatus { + /// Required work completed. + Succeeded, + /// Optional work failed and startup may continue. + Degraded, + /// Required work failed. + Failed, + /// Control flow dropped the phase without an explicit terminal. + Abandoned, +} + +impl LifecycleStatus { + #[cfg(test)] + const ALL: [Self; 4] = [ + Self::Succeeded, + Self::Degraded, + Self::Failed, + Self::Abandoned, + ]; + + const fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "succeeded", + Self::Degraded => "degraded", + Self::Failed => "failed", + Self::Abandoned => "abandoned", + } + } +} + +/// A secret-safe terminal reason. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum LifecycleReason { + /// Tokio runtime construction failed. + RuntimeBuild, + /// Another rustls provider was already installed. + ProviderConflict, + /// The optional OTLP exporter could not be built. + ExporterBuild, + /// Required configuration was missing, malformed, or unusable. + ConfigInvalid, + /// A required value was missing. + Missing, + /// A required value was invalid. + RequiredInvalid, + /// A required listener could not bind. + Bind, + /// A global metrics recorder already existed. + RecorderConflict, + /// A phase owner disappeared without a terminal. + OwnerDropped, + /// A panic unwound through the phase. + Panic, +} + +impl LifecycleReason { + #[cfg(test)] + const ALL: [Self; 10] = [ + Self::RuntimeBuild, + Self::ProviderConflict, + Self::ExporterBuild, + Self::ConfigInvalid, + Self::Missing, + Self::RequiredInvalid, + Self::Bind, + Self::RecorderConflict, + Self::OwnerDropped, + Self::Panic, + ]; + + /// Stable wire value. + pub const fn as_str(self) -> &'static str { + match self { + Self::RuntimeBuild => "runtime_build", + Self::ProviderConflict => "provider_conflict", + Self::ExporterBuild => "exporter_build", + Self::ConfigInvalid => "config_invalid", + Self::Missing => "missing", + Self::RequiredInvalid => "required_invalid", + Self::Bind => "bind", + Self::RecorderConflict => "recorder_conflict", + Self::OwnerDropped => "owner_dropped", + Self::Panic => "panic", + } + } +} + +#[derive(Clone, Debug, Serialize)] +struct LifecycleEvent { + event_name: &'static str, + schema_version: u8, + process_boot_id: Uuid, + sequence: u64, + track: &'static str, + phase: &'static str, + edge: &'static str, + #[serde(skip_serializing_if = "Option::is_none")] + status: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + reason: Option<&'static str>, + process_started_at_unix_ms: u64, + observed_at_unix_ms: u64, + process_elapsed_ms: u64, + #[serde(skip_serializing_if = "Option::is_none")] + phase_elapsed_ms: Option, +} + +trait EventWriter: Send + Sync { + fn emit(&self, event: &LifecycleEvent); +} + +struct StderrWriter; + +impl EventWriter for StderrWriter { + fn emit(&self, event: &LifecycleEvent) { + // Best effort: reporting a startup error must never create another + // panic. This sink intentionally ignores RUST_LOG filters. + let mut stderr = std::io::stderr().lock(); + if serde_json::to_writer(&mut stderr, event).is_ok() { + let _ = stderr.write_all(b"\n"); + } + } +} + +struct ProcessLifecycle { + boot_id: Uuid, + sequence: AtomicU64, + wall_origin: SystemTime, + monotonic_origin: Instant, + writer: Arc, +} + +impl ProcessLifecycle { + fn new(writer: Arc) -> Arc { + let wall_origin = SystemTime::now(); + let monotonic_origin = Instant::now(); + Arc::new(Self { + boot_id: Uuid::new_v4(), + sequence: AtomicU64::new(1), + wall_origin, + monotonic_origin, + writer, + }) + } + + fn start(self: &Arc, phase: StartupPhase) -> PhaseGuard { + let started_at = if phase == StartupPhase::ProcessTelemetry { + self.monotonic_origin + } else { + Instant::now() + }; + self.emit(phase, "started", None, None, None); + PhaseGuard { + lifecycle: Arc::clone(self), + phase, + started_at, + finished: false, + } + } + + fn emit( + &self, + phase: StartupPhase, + edge: &'static str, + status: Option, + reason: Option, + elapsed: Option, + ) { + self.writer.emit(&LifecycleEvent { + event_name: EVENT_NAME, + schema_version: SCHEMA_VERSION, + process_boot_id: self.boot_id, + sequence: self.sequence.fetch_add(1, Ordering::Relaxed), + track: "startup", + phase: phase.as_str(), + edge, + status: status.map(LifecycleStatus::as_str), + reason: reason.map(LifecycleReason::as_str), + process_started_at_unix_ms: millis_since_epoch(self.wall_origin), + observed_at_unix_ms: millis_since_epoch(SystemTime::now()), + process_elapsed_ms: saturating_millis(self.monotonic_origin.elapsed()), + phase_elapsed_ms: elapsed.map(saturating_millis), + }); + } +} + +/// Owns one phase from its start event through exactly one terminal. +pub struct PhaseGuard { + lifecycle: Arc, + phase: StartupPhase, + started_at: Instant, + finished: bool, +} + +impl PhaseGuard { + /// Record successful completion. + pub fn succeed(self) { + self.finish(LifecycleStatus::Succeeded, None); + } + + /// Record an allowed degradation. + pub fn degrade(self, reason: LifecycleReason) { + self.finish(LifecycleStatus::Degraded, Some(reason)); + } + + /// Record a fatal failure. + pub fn fail(self, reason: LifecycleReason) { + self.finish(LifecycleStatus::Failed, Some(reason)); + } + + fn finish(mut self, status: LifecycleStatus, reason: Option) { + let elapsed = self.started_at.elapsed(); + self.lifecycle + .emit(self.phase, "terminal", Some(status), reason, Some(elapsed)); + self.finished = true; + } +} + +impl Drop for PhaseGuard { + fn drop(&mut self) { + if self.finished { + return; + } + let (status, reason) = if std::thread::panicking() { + (LifecycleStatus::Failed, LifecycleReason::Panic) + } else { + (LifecycleStatus::Abandoned, LifecycleReason::OwnerDropped) + }; + self.lifecycle.emit( + self.phase, + "terminal", + Some(status), + Some(reason), + Some(self.started_at.elapsed()), + ); + self.finished = true; + } +} + +/// Tracks the aggregate early-startup phase and its fixed subphases. +pub struct BootTracker { + lifecycle: Arc, + headline: PhaseGuard, + degraded: Option, +} + +impl BootTracker { + /// Start lifecycle accounting before constructing Tokio. + pub fn start_before_runtime( + build: impl FnOnce() -> Result, + ) -> Result<(Runtime, Self), Error> { + Self::start_before_runtime_with_writer(Arc::new(StderrWriter), build) + } + + fn start_before_runtime_with_writer( + writer: Arc, + build: impl FnOnce() -> Result, + ) -> Result<(Runtime, Self), Error> { + let lifecycle = ProcessLifecycle::new(writer); + let boot = Self { + headline: lifecycle.start(StartupPhase::ProcessTelemetry), + lifecycle, + degraded: None, + }; + match build() { + Ok(runtime) => Ok((runtime, boot)), + Err(error) => { + boot.fail(LifecycleReason::RuntimeBuild); + Err(error) + } + } + } + + /// Start a fixed early-startup subphase. + #[must_use = "dropping a phase guard emits an abandoned terminal"] + pub fn start(&self, phase: StartupPhase) -> PhaseGuard { + assert_ne!(phase, StartupPhase::ProcessTelemetry); + self.lifecycle.start(phase) + } + + /// Run a required phase and atomically terminalize both it and startup on failure. + pub fn run_required( + self, + phase: StartupPhase, + work: impl FnOnce() -> Result, + classify: impl FnOnce(&Error) -> LifecycleReason, + ) -> Result<(Self, T), Error> { + let phase_guard = self.start(phase); + match work() { + Ok(value) => { + phase_guard.succeed(); + Ok((self, value)) + } + Err(error) => { + let reason = classify(&error); + phase_guard.fail(reason); + self.fail(reason); + Err(error) + } + } + } + + /// Preserve the first optional degradation for the aggregate terminal. + pub fn mark_degraded(&mut self, reason: LifecycleReason) { + self.degraded.get_or_insert(reason); + } + + /// Finish early startup with a structured lifecycle terminal. + pub fn finish(self) { + let status = if self.degraded.is_some() { + LifecycleStatus::Degraded + } else { + LifecycleStatus::Succeeded + }; + self.headline.finish(status, self.degraded); + } + + fn fail(self, reason: LifecycleReason) { + self.headline.fail(reason); + } +} + +fn millis_since_epoch(time: SystemTime) -> u64 { + time.duration_since(UNIX_EPOCH) + .map(saturating_millis) + .unwrap_or(0) +} + +fn saturating_millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{panic::AssertUnwindSafe, sync::Mutex}; + + #[derive(Default)] + struct CapturingWriter(Mutex>); + + impl EventWriter for CapturingWriter { + fn emit(&self, event: &LifecycleEvent) { + self.0.lock().expect("capturing writer").push(event.clone()); + } + } + + fn recorder() -> (Arc, Arc) { + let writer = Arc::new(CapturingWriter::default()); + (ProcessLifecycle::new(writer.clone()), writer) + } + + fn events(writer: &CapturingWriter) -> Vec { + writer.0.lock().expect("capturing writer").clone() + } + + #[test] + fn explicit_and_dropped_terminals_are_exactly_once() { + let (lifecycle, writer) = recorder(); + lifecycle.start(StartupPhase::ConfigLoad).succeed(); + drop(lifecycle.start(StartupPhase::KeyLoad)); + + let events = events(&writer); + assert_eq!(events.len(), 4); + assert_eq!(events[0].sequence, 1); + assert_eq!(events[1].status, Some("succeeded")); + assert_eq!(events[3].status, Some("abandoned")); + assert_eq!(events[3].reason, Some("owner_dropped")); + } + + #[test] + fn panic_unwind_is_bounded() { + let (lifecycle, writer) = recorder(); + let panic = std::panic::catch_unwind(AssertUnwindSafe(|| { + let _phase = lifecycle.start(StartupPhase::CryptoInit); + panic!("controlled test panic"); + })); + assert!(panic.is_err()); + let events = events(&writer); + assert_eq!(events[1].status, Some("failed")); + assert_eq!(events[1].reason, Some("panic")); + } + + #[test] + fn runtime_failure_terminalizes_the_headline() { + let writer = Arc::new(CapturingWriter::default()); + let result = BootTracker::start_before_runtime_with_writer( + writer.clone(), + || -> Result<(), &'static str> { Err("controlled") }, + ); + assert!(matches!(result, Err("controlled"))); + let events = events(&writer); + assert_eq!(events.len(), 2); + assert_eq!(events[1].phase, "process_telemetry"); + assert_eq!(events[1].reason, Some("runtime_build")); + } + + #[test] + fn aggregate_preserves_optional_degradation() { + let (lifecycle, writer) = recorder(); + let mut boot = BootTracker { + headline: lifecycle.start(StartupPhase::ProcessTelemetry), + lifecycle, + degraded: None, + }; + boot.mark_degraded(LifecycleReason::ExporterBuild); + boot.finish(); + let events = events(&writer); + assert_eq!(events[1].status, Some("degraded")); + assert_eq!(events[1].reason, Some("exporter_build")); + } + + #[test] + fn required_failure_terminalizes_subphase_and_headline() { + let (lifecycle, writer) = recorder(); + let boot = BootTracker { + headline: lifecycle.start(StartupPhase::ProcessTelemetry), + lifecycle, + degraded: None, + }; + let result = boot.run_required( + StartupPhase::MetricsBind, + || -> Result<(), &'static str> { Err("controlled") }, + |_error| LifecycleReason::RecorderConflict, + ); + assert!(matches!(result, Err("controlled"))); + + let events = events(&writer); + assert_eq!(events.len(), 4); + assert_eq!(events[2].phase, "metrics_bind"); + assert_eq!(events[2].status, Some("failed")); + assert_eq!(events[2].reason, Some("recorder_conflict")); + assert_eq!(events[3].phase, "process_telemetry"); + assert_eq!(events[3].status, Some("failed")); + assert_eq!(events[3].reason, Some("recorder_conflict")); + } + + #[test] + fn schema_and_vocabulary_are_frozen() { + assert_eq!( + StartupPhase::ALL.map(StartupPhase::as_str), + [ + "process_telemetry", + "crypto_init", + "tracing_init", + "config_load", + "key_load", + "metrics_bind", + ] + ); + let (lifecycle, writer) = recorder(); + drop(lifecycle.start(StartupPhase::ConfigLoad)); + let values: Vec<_> = events(&writer) + .iter() + .map(|event| serde_json::to_value(event).expect("serialize lifecycle event")) + .collect(); + assert_eq!(values[0]["schema_version"], SCHEMA_VERSION); + assert_eq!(values[0]["event_name"], EVENT_NAME); + assert_eq!(values[1]["status"], "abandoned"); + let mut started_keys: Vec<_> = values[0] + .as_object() + .expect("started event object") + .keys() + .map(String::as_str) + .collect(); + started_keys.sort_unstable(); + assert_eq!( + started_keys, + [ + "edge", + "event_name", + "observed_at_unix_ms", + "phase", + "process_boot_id", + "process_elapsed_ms", + "process_started_at_unix_ms", + "schema_version", + "sequence", + "track", + ] + ); + let mut terminal_keys: Vec<_> = values[1] + .as_object() + .expect("terminal event object") + .keys() + .map(String::as_str) + .collect(); + terminal_keys.sort_unstable(); + assert_eq!( + terminal_keys, + [ + "edge", + "event_name", + "observed_at_unix_ms", + "phase", + "phase_elapsed_ms", + "process_boot_id", + "process_elapsed_ms", + "process_started_at_unix_ms", + "reason", + "schema_version", + "sequence", + "status", + "track", + ] + ); + assert_eq!( + LifecycleStatus::ALL.map(LifecycleStatus::as_str), + ["succeeded", "degraded", "failed", "abandoned",] + ); + assert_eq!( + LifecycleReason::ALL.map(LifecycleReason::as_str), + [ + "runtime_build", + "provider_conflict", + "exporter_build", + "config_invalid", + "missing", + "required_invalid", + "bind", + "recorder_conflict", + "owner_dropped", + "panic", + ] + ); + } +} diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 933756aa106..206f0329c0e 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -17,6 +17,7 @@ use buzz_pubsub::PubSubManager; use buzz_search::SearchService; use buzz_relay::config::{Config, MAX_DRAIN_JITTER_MS}; +use buzz_relay::lifecycle::{BootTracker, LifecycleReason, StartupPhase}; use buzz_relay::metrics as relay_metrics; use buzz_relay::router::{build_health_router, build_router}; use buzz_relay::state::AppState; @@ -104,15 +105,36 @@ impl EmissionScope { const USAGE_METRICS_LOCK_KEY: i64 = 0x4255_5A5A_4D45_5452; -#[tokio::main] -async fn main() -> anyhow::Result<()> { +fn main() -> anyhow::Result<()> { + let (runtime, boot) = BootTracker::start_before_runtime(|| { + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + }) + .map_err(|error| anyhow::anyhow!("failed to build Tokio runtime: {error}"))?; + runtime.block_on(run_relay_main(boot)) +} + +async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { // Install the ring CryptoProvider for rustls. Required before any rustls // TLS connection (rediss:// to ElastiCache, wss://, S3 over TLS): both // aws-lc-rs and ring are compiled in transitively, so rustls can't // auto-select a provider and would panic at first use without this. - rustls::crypto::ring::default_provider() - .install_default() - .expect("failed to install rustls crypto provider"); + let (mut boot, ()) = boot + .run_required( + StartupPhase::CryptoInit, + || { + rustls::crypto::ring::default_provider() + .install_default() + .map_err(|_provider| ()) + }, + |_error| LifecycleReason::ProviderConflict, + ) + .map_err(|()| { + anyhow::anyhow!( + "failed to install rustls crypto provider: another provider is already installed" + ) + })?; // JSON-only structured logs — simple, machine-parseable, CAKE-compatible. // If OTEL_EXPORTER_OTLP_ENDPOINT is set, also attach an OpenTelemetry tracing @@ -121,6 +143,7 @@ async fn main() -> anyhow::Result<()> { // Build a single shared Resource (service.name=buzz-relay by default, overridable // via OTEL_SERVICE_NAME) for the trace provider so that Datadog can identify // spans under the correct service identity. + let tracing_init = boot.start(StartupPhase::TracingInit); let resource = telemetry::service_resource(); let tracer_init = telemetry::try_init_tracer(resource.clone()); let otel_enabled = matches!(&tracer_init, telemetry::TracerInit::Enabled(_)); @@ -154,17 +177,43 @@ async fn main() -> anyhow::Result<()> { .init(); // Log any exporter-build failure now that the subscriber is installed. - if let telemetry::TracerInit::ExporterBuildFailed(ref e) = tracer_init { - warn!(error = %e, "Failed to build OTLP trace exporter; distributed tracing disabled"); + match &tracer_init { + telemetry::TracerInit::Enabled(_) => tracing_init.succeed(), + // Structured logging is installed regardless of whether optional OTLP + // export is configured, so the phase itself completed successfully. + telemetry::TracerInit::Disabled => tracing_init.succeed(), + telemetry::TracerInit::ExporterBuildFailed(_) => { + tracing_init.degrade(LifecycleReason::ExporterBuild); + boot.mark_degraded(LifecycleReason::ExporterBuild); + // Do not log the raw exporter error: OTLP endpoint URLs can carry + // credentials. The bounded lifecycle reason is sufficient here. + warn!("Failed to build OTLP trace exporter; distributed tracing disabled"); + } } info!("Starting buzz-relay"); - let config = Config::from_env().map_err(|e| { - error!("Invalid configuration: {e}"); - anyhow::anyhow!("Configuration error: {e}") - })?; - let relay_keypair = relay_keypair_from_config(config.relay_private_key.as_deref())?; + let (next_boot, config) = boot + .run_required(StartupPhase::ConfigLoad, Config::from_env, |_error| { + LifecycleReason::ConfigInvalid + }) + .map_err(|error| { + error!("Invalid configuration: {error}"); + anyhow::anyhow!("Configuration error: {error}") + })?; + boot = next_boot; + + let key_failure = if config.relay_private_key.is_some() { + LifecycleReason::RequiredInvalid + } else { + LifecycleReason::Missing + }; + let (next_boot, relay_keypair) = boot.run_required( + StartupPhase::KeyLoad, + || relay_keypair_from_config(config.relay_private_key.as_deref()), + |_error| key_failure, + )?; + boot = next_boot; info!( bind_addr = %config.bind_addr, relay_url = %config.relay_url, @@ -178,7 +227,18 @@ async fn main() -> anyhow::Result<()> { let usage_interval_secs = usage_metrics_interval_secs(); let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(usage_interval_secs); - relay_metrics::install(config.metrics_port, usage_idle_timeout_secs); + let (boot, ()) = boot.run_required( + StartupPhase::MetricsBind, + || relay_metrics::try_install(config.metrics_port, usage_idle_timeout_secs), + |error| match error.failure() { + relay_metrics::MetricsInstallFailure::Bind => LifecycleReason::Bind, + relay_metrics::MetricsInstallFailure::RecorderConflict => { + LifecycleReason::RecorderConflict + } + relay_metrics::MetricsInstallFailure::ExporterBuild => LifecycleReason::ExporterBuild, + }, + )?; + boot.finish(); metrics::gauge!("buzz_audit_enabled").set(if config.audit_enabled { 1.0 } else { 0.0 }); metrics::gauge!("buzz_push_enabled").set(if config.push_enabled { 1.0 } else { 0.0 }); info!( diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 0c4bfd2c31f..f71894116c3 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -21,7 +21,7 @@ use axum::{ middleware::Next, response::Response, }; -use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; +use metrics_exporter_prometheus::{BuildError, Matcher, PrometheusBuilder}; use metrics_util::MetricKindMask; /// HTTP latency buckets (milliseconds) — only for `http_request_latency_ms`. @@ -154,23 +154,69 @@ fn configured_prometheus_builder(gauge_idle_timeout_secs: u64) -> PrometheusBuil .expect("valid fanout bucket boundaries") } -/// Install the global metrics recorder and spawn the Prometheus HTTP exporter. +/// A bounded class of metrics installation failure. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum MetricsInstallFailure { + /// The Prometheus listener could not bind. + Bind, + /// Another component already installed a global recorder. + RecorderConflict, + /// The exporter could not be built for another reason. + ExporterBuild, +} + +/// An error returned while installing Prometheus metrics. +#[derive(Debug, thiserror::Error)] +pub enum MetricsInstallError { + /// Prometheus exporter construction failed. + #[error("failed to build Prometheus exporter: {0}")] + Build(#[source] BuildError), + /// Another component already installed the process-global recorder. + #[error("the global metrics recorder is already installed")] + RecorderConflict, +} + +impl MetricsInstallError { + /// Return the secret-safe lifecycle classification. + pub const fn failure(&self) -> MetricsInstallFailure { + match self { + Self::Build(BuildError::FailedToCreateHTTPListener(_)) => MetricsInstallFailure::Bind, + Self::Build(_) => MetricsInstallFailure::ExporterBuild, + Self::RecorderConflict => MetricsInstallFailure::RecorderConflict, + } + } +} + +/// Try to install the global metrics recorder and spawn the Prometheus HTTP exporter. /// /// `build()` returns the recorder + exporter future and internally spawns /// the upkeep task, so no separate upkeep call is needed. /// /// Must be called from within a Tokio runtime. -/// Panics if a recorder is already installed or the port is in use. -pub fn install(port: u16, gauge_idle_timeout_secs: u64) { +/// Listener and global-recorder failures are returned rather than panicking. +/// A later exporter exit remains detached from relay service; external scrape +/// coverage is authoritative for exporter availability. +pub fn try_install(port: u16, gauge_idle_timeout_secs: u64) -> Result<(), MetricsInstallError> { let (recorder, exporter) = configured_prometheus_builder(gauge_idle_timeout_secs) .with_http_listener(([0, 0, 0, 0], port)) .build() - .expect("metrics exporter must build exactly once"); + .map_err(MetricsInstallError::Build)?; - metrics::set_global_recorder(recorder).expect("global recorder must be set exactly once"); + metrics::set_global_recorder(recorder) + .map_err(|_error| MetricsInstallError::RecorderConflict)?; describe_readiness_metrics(); describe_db_pool_metrics(); tokio::spawn(exporter); + Ok(()) +} + +/// Install the global metrics recorder and spawn the Prometheus HTTP exporter. +/// +/// This compatibility entry point preserves the original panic-on-failure API. +/// New startup code should use [`try_install`] to report typed failures. +pub fn install(port: u16, gauge_idle_timeout_secs: u64) { + try_install(port, gauge_idle_timeout_secs) + .unwrap_or_else(|error| panic!("metrics exporter must install exactly once: {error}")); } /// Register the frozen readiness metric descriptions with the active recorder. @@ -280,7 +326,6 @@ pub async fn track_metrics(req: Request, next: Next) -> Response { response } - #[cfg(test)] mod contract_tests { use std::collections::BTreeSet; @@ -387,3 +432,33 @@ mod contract_tests { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn occupied_listener_is_classified_as_bind() { + let listener = std::net::TcpListener::bind(("0.0.0.0", 0)).expect("bind occupied port"); + let port = listener.local_addr().expect("occupied address").port(); + let error = try_install(port, 300).expect_err("occupied listener must fail"); + assert_eq!(error.failure(), MetricsInstallFailure::Bind); + } + + #[tokio::test] + async fn recorder_conflict_is_typed_in_an_isolated_process() { + const CHILD_ENV: &str = "BUZZ_TEST_METRICS_RECORDER_CONFLICT"; + if std::env::var_os(CHILD_ENV).is_some() { + let recorder = configured_prometheus_builder(300).build_recorder(); + metrics::set_global_recorder(recorder).expect("install first recorder"); + let error = try_install(0, 300).expect_err("second recorder must fail"); + assert_eq!(error.failure(), MetricsInstallFailure::RecorderConflict); + return; + } + + crate::test_support::run_exact_test_child( + "metrics::tests::recorder_conflict_is_typed_in_an_isolated_process", + CHILD_ENV, + ); + } +} diff --git a/crates/buzz-relay/src/telemetry.rs b/crates/buzz-relay/src/telemetry.rs index 91bd92f0f3e..7ffd6a7330b 100644 --- a/crates/buzz-relay/src/telemetry.rs +++ b/crates/buzz-relay/src/telemetry.rs @@ -223,9 +223,9 @@ pub enum TracerInit { Enabled(SdkTracerProvider), /// `OTEL_EXPORTER_OTLP_ENDPOINT` was unset — no-op, no connection. Disabled, - /// Endpoint was set but the exporter failed to build. The inner error - /// string is suitable for a `tracing::warn!` call made by the caller - /// **after** `tracing_subscriber::registry()…init()`. + /// Endpoint was set but the exporter failed to build. The inner error is + /// diagnostic data only and must not be logged: exporter errors can + /// include credential-bearing endpoint URLs. ExporterBuildFailed(String), } @@ -234,7 +234,8 @@ pub enum TracerInit { /// /// Deliberately does **not** call `tracing::warn!` internally — the subscriber /// may not be installed yet at call time, which would silently drop the event. -/// Callers are responsible for logging [`TracerInit::ExporterBuildFailed`]. +/// Callers may log a fixed, credential-free message for +/// [`TracerInit::ExporterBuildFailed`], but must not log its inner error. pub fn try_init_tracer(resource: Resource) -> TracerInit { if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_err() { return TracerInit::Disabled; diff --git a/crates/buzz-relay/src/test_support.rs b/crates/buzz-relay/src/test_support.rs index 6936a60ae4e..a0b9f685374 100644 --- a/crates/buzz-relay/src/test_support.rs +++ b/crates/buzz-relay/src/test_support.rs @@ -7,3 +7,102 @@ pub(crate) fn database_url() -> String { .or_else(|_| std::env::var("DATABASE_URL")) .unwrap_or_else(|_| DEFAULT_DATABASE_URL.to_owned()) } + +#[cfg(test)] +const CHILD_TEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +#[cfg(test)] +const MAX_CAPTURE_BYTES: u64 = 1024 * 1024; + +#[cfg(test)] +struct CapturedStream { + retained: Vec, + total_bytes: u64, +} + +#[cfg(test)] +fn capture_stream(mut stream: impl std::io::Read) -> CapturedStream { + let mut retained = Vec::new(); + let mut total_bytes = 0_u64; + let mut chunk = [0_u8; 8192]; + loop { + let read = stream.read(&mut chunk).expect("read child output pipe"); + if read == 0 { + break; + } + total_bytes = total_bytes.saturating_add(u64::try_from(read).expect("read size fits u64")); + let remaining = usize::try_from(MAX_CAPTURE_BYTES) + .expect("capture ceiling fits usize") + .saturating_sub(retained.len()); + retained.extend_from_slice(&chunk[..read.min(remaining)]); + } + CapturedStream { + retained, + total_bytes, + } +} + +#[cfg(test)] +fn join_capture(capture: std::thread::JoinHandle, stream: &str) -> Vec { + let capture = capture.join().expect("child capture thread must not panic"); + assert!( + capture.total_bytes <= MAX_CAPTURE_BYTES, + "child {stream} exceeded {MAX_CAPTURE_BYTES} bytes: {}", + capture.total_bytes, + ); + capture.retained +} + +/// Run exactly one unit test in an isolated, deadline-bounded child process. +#[cfg(test)] +pub(crate) fn run_exact_test_child(test_name: &str, child_env: &str) { + use std::{ + process::{Command, Stdio}, + thread, + time::{Duration, Instant}, + }; + + let mut child = Command::new(std::env::current_exe().expect("test executable")) + .arg("--exact") + .arg(test_name) + .arg("--nocapture") + .env(child_env, "1") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn isolated test child"); + let stdout = child.stdout.take().expect("child stdout pipe"); + let stderr = child.stderr.take().expect("child stderr pipe"); + let stdout = thread::spawn(move || capture_stream(stdout)); + let stderr = thread::spawn(move || capture_stream(stderr)); + + let deadline = Instant::now() + CHILD_TEST_TIMEOUT; + let (status, timed_out) = loop { + if let Some(status) = child.try_wait().expect("poll isolated test child") { + break (status, false); + } + if Instant::now() >= deadline { + let _ = child.kill(); + let status = child.wait().expect("reap timed-out test child"); + break (status, true); + } + thread::sleep(Duration::from_millis(10)); + }; + + let stdout = join_capture(stdout, "stdout"); + let stderr = join_capture(stderr, "stderr"); + let output = format!( + "{}{}", + String::from_utf8_lossy(&stdout), + String::from_utf8_lossy(&stderr) + ); + + assert!( + !timed_out, + "isolated test child exceeded {CHILD_TEST_TIMEOUT:?}:\n{output}" + ); + assert!(status.success(), "isolated test child failed:\n{output}"); + assert!( + output.contains("running 1 test") && output.contains(test_name), + "exact selector did not run the intended test {test_name}:\n{output}" + ); +} diff --git a/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs new file mode 100644 index 00000000000..29fcf991f8d --- /dev/null +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -0,0 +1,457 @@ +use std::{ + collections::BTreeMap, + io::{Read as _, Write as _}, + net::{TcpListener, TcpStream}, + process::{Child, Command, ExitStatus, Output, Stdio}, + thread::{self, JoinHandle}, + time::{Duration, Instant}, +}; + +use serde_json::Value; + +use buzz_relay::lifecycle::StartupPhase; + +const VALID_RELAY_PRIVATE_KEY: &str = + "0000000000000000000000000000000000000000000000000000000000000001"; +const CHILD_TIMEOUT: Duration = Duration::from_secs(10); +const MAX_CAPTURE_BYTES: u64 = 1024 * 1024; + +struct RelayProcess { + child: Option, + stdout: Option>, + stderr: Option>, + scratch_dir: std::path::PathBuf, +} + +struct CapturedStream { + retained: Vec, + total_bytes: u64, +} + +impl RelayProcess { + fn spawn(environment: &[(&str, &str)]) -> Self { + let scratch_dir = + std::env::temp_dir().join(format!("buzz-boot-lifecycle-{}", uuid::Uuid::new_v4())); + let mut command = Command::new(env!("CARGO_BIN_EXE_buzz-relay")); + command + .env_clear() + .env("RUST_BACKTRACE", "0") + .env("RUST_LOG", "buzz_relay=info") + .env("BUZZ_GIT_REPO_PATH", scratch_dir.join("repos")) + .env("BUZZ_GIT_PACK_CACHE_PATH", scratch_dir.join("pack-cache")) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + for (name, value) in environment { + command.env(name, value); + } + let mut child = command.spawn().expect("spawn buzz-relay child process"); + let stdout = child.stdout.take().expect("relay stdout pipe"); + let stderr = child.stderr.take().expect("relay stderr pipe"); + Self { + child: Some(child), + stdout: Some(thread::spawn(move || capture_stream(stdout))), + stderr: Some(thread::spawn(move || capture_stream(stderr))), + scratch_dir, + } + } + + fn try_wait(&mut self) -> Option { + self.child + .as_mut() + .expect("relay child") + .try_wait() + .expect("poll relay child") + } + + fn wait(mut self, timeout: Duration) -> Output { + let deadline = Instant::now() + timeout; + let status = loop { + if let Some(status) = self.try_wait() { + break status; + } + if Instant::now() >= deadline { + let child = self.child.as_mut().expect("relay child"); + let _ = child.kill(); + let _ = child.wait(); + panic!("buzz-relay child exceeded {timeout:?}"); + } + thread::sleep(Duration::from_millis(10)); + }; + self.child.take(); + let output = Output { + status, + stdout: join_capture(self.stdout.take(), "stdout"), + stderr: join_capture(self.stderr.take(), "stderr"), + }; + let _ = std::fs::remove_dir_all(&self.scratch_dir); + output + } + + fn terminate(mut self) -> Output { + self.child + .as_mut() + .expect("relay child") + .kill() + .expect("terminate exact relay child"); + self.wait(Duration::from_secs(2)) + } +} + +impl Drop for RelayProcess { + fn drop(&mut self) { + if let Some(child) = self.child.as_mut() { + let _ = child.kill(); + let _ = child.wait(); + } + let _ = std::fs::remove_dir_all(&self.scratch_dir); + } +} + +fn capture_stream(mut stream: impl std::io::Read) -> CapturedStream { + let mut retained = Vec::new(); + let mut total_bytes = 0_u64; + let mut chunk = [0_u8; 8192]; + loop { + let read = stream.read(&mut chunk).expect("read relay output pipe"); + if read == 0 { + break; + } + total_bytes = total_bytes.saturating_add(u64::try_from(read).expect("read size fits u64")); + let remaining = usize::try_from(MAX_CAPTURE_BYTES) + .expect("capture ceiling fits usize") + .saturating_sub(retained.len()); + retained.extend_from_slice(&chunk[..read.min(remaining)]); + } + CapturedStream { + retained, + total_bytes, + } +} + +fn join_capture(capture: Option>, stream: &str) -> Vec { + let capture = capture + .expect("relay capture thread") + .join() + .expect("relay capture thread must not panic"); + assert!( + capture.total_bytes <= MAX_CAPTURE_BYTES, + "relay {stream} exceeded {MAX_CAPTURE_BYTES} bytes: {}", + capture.total_bytes, + ); + capture.retained +} + +fn run_relay(environment: &[(&str, &str)]) -> Output { + RelayProcess::spawn(environment).wait(CHILD_TIMEOUT) +} + +fn scrape_metrics(port: u16) -> std::io::Result { + let address = std::net::SocketAddr::from(([127, 0, 0, 1], port)); + let mut stream = TcpStream::connect_timeout(&address, Duration::from_millis(100))?; + stream.set_read_timeout(Some(Duration::from_millis(500)))?; + stream.write_all(b"GET /metrics HTTP/1.0\r\nHost: 127.0.0.1\r\n\r\n")?; + let mut response = String::new(); + stream.read_to_string(&mut response)?; + Ok(response) +} + +fn wait_for_relay_metrics(process: &mut RelayProcess, port: u16) -> String { + let deadline = Instant::now() + Duration::from_secs(8); + loop { + assert!( + process.try_wait().is_none(), + "relay exited before its metrics endpoint became usable" + ); + if let Ok(response) = scrape_metrics(port) { + if response.contains("buzz_audit_enabled") { + return response; + } + } + assert!( + Instant::now() < deadline, + "relay metrics did not become scrapeable within 8s" + ); + thread::sleep(Duration::from_millis(20)); + } +} + +fn assert_no_startup_lifecycle_metrics(scrape: &str) { + for line in scrape.lines() { + let Some(name) = line + .strip_prefix("# HELP ") + .or_else(|| line.strip_prefix("# TYPE ")) + .and_then(|rest| rest.split_ascii_whitespace().next()) + else { + continue; + }; + assert!( + !["startup", "boot", "lifecycle"] + .iter() + .any(|term| name.contains(term)) + && !StartupPhase::ALL + .iter() + .any(|phase| name.contains(phase.as_str())), + "logs-only lifecycle contract emitted metric family {name}" + ); + } +} + +fn lifecycle_events(output: &Output) -> Vec { + let mut events: Vec = output + .stdout + .split(|byte| *byte == b'\n') + .chain(output.stderr.split(|byte| *byte == b'\n')) + .filter_map(|line| serde_json::from_slice::(line).ok()) + .filter(|event| event["event_name"] == "buzz_process_lifecycle") + .collect(); + events.sort_by_key(|event| event["sequence"].as_u64()); + events +} + +fn lifecycle_events_from(bytes: &[u8]) -> Vec { + bytes + .split(|byte| *byte == b'\n') + .filter_map(|line| serde_json::from_slice::(line).ok()) + .filter(|event| event["event_name"] == "buzz_process_lifecycle") + .collect() +} + +fn assert_accounting(events: &[Value]) { + assert!(!events.is_empty(), "child emitted no lifecycle events"); + let boot_id = events[0]["process_boot_id"] + .as_str() + .expect("process_boot_id"); + let mut counts = BTreeMap::::new(); + for (index, event) in events.iter().enumerate() { + assert_eq!(event["schema_version"], 1); + assert_eq!(event["sequence"], u64::try_from(index + 1).unwrap()); + assert_eq!(event["process_boot_id"], boot_id); + assert_eq!(event["track"], "startup"); + let count = counts + .entry(event["phase"].as_str().expect("phase").to_owned()) + .or_default(); + match event["edge"].as_str() { + Some("started") => count.0 += 1, + Some("terminal") => count.1 += 1, + other => panic!("unexpected lifecycle edge: {other:?}"), + } + } + assert!( + counts + .values() + .all(|(started, terminal)| *started == 1 && *terminal == 1), + "every started phase must have one terminal: {counts:?}" + ); +} + +fn assert_terminal(events: &[Value], phase: &str, status: &str, reason: Option<&str>) { + let terminal = events + .iter() + .find(|event| event["phase"] == phase && event["edge"] == "terminal") + .unwrap_or_else(|| panic!("missing {phase} terminal")); + assert_eq!(terminal["status"], status); + match reason { + Some(reason) => assert_eq!(terminal["reason"], reason), + None => assert!(terminal["reason"].is_null()), + } +} + +fn phases(events: &[Value]) -> Vec<&str> { + events + .iter() + .filter(|event| event["edge"] == "started") + .map(|event| event["phase"].as_str().expect("phase")) + .collect() +} + +#[test] +fn invalid_config_terminalizes_at_main_even_with_logs_disabled() { + let output = run_relay(&[ + ("RUST_LOG", "off"), + ("BUZZ_BIND_ADDR", "not-a-socket-address"), + ]); + assert!(!output.status.success()); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_eq!( + phases(&events), + [ + "process_telemetry", + "crypto_init", + "tracing_init", + "config_load" + ] + ); + assert_terminal(&events, "config_load", "failed", Some("config_invalid")); + assert_terminal( + &events, + "process_telemetry", + "failed", + Some("config_invalid"), + ); + assert_eq!(lifecycle_events_from(&output.stderr), events); + assert!(lifecycle_events_from(&output.stdout).is_empty()); +} + +#[test] +#[cfg(unix)] +fn config_filesystem_failure_has_a_bounded_terminal() { + let output = run_relay(&[ + ("RUST_LOG", "off"), + ("BUZZ_GIT_REPO_PATH", "/dev/null/not-a-directory"), + ]); + assert!(!output.status.success()); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_terminal(&events, "config_load", "failed", Some("config_invalid")); + assert_terminal( + &events, + "process_telemetry", + "failed", + Some("config_invalid"), + ); +} + +#[test] +fn invalid_config_value_has_the_same_bounded_terminal() { + let output = run_relay(&[("RUST_LOG", "off"), ("BUZZ_DRAIN_JITTER_MS", "bogus")]); + assert!(!output.status.success()); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_terminal(&events, "config_load", "failed", Some("config_invalid")); + assert_terminal( + &events, + "process_telemetry", + "failed", + Some("config_invalid"), + ); +} + +#[test] +fn configured_otlp_terminalizes_tracing_before_a_later_failure() { + let output = run_relay(&[ + ("RUST_LOG", "off"), + ("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:4317"), + ("BUZZ_BIND_ADDR", "not-a-socket-address"), + ]); + assert!(!output.status.success()); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_terminal(&events, "tracing_init", "succeeded", None); + assert_terminal(&events, "config_load", "failed", Some("config_invalid")); +} + +#[test] +fn missing_key_stops_before_metrics_bind() { + let output = run_relay(&[]); + assert!(!output.status.success()); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_eq!( + phases(&events), + [ + "process_telemetry", + "crypto_init", + "tracing_init", + "config_load", + "key_load" + ] + ); + assert_terminal(&events, "key_load", "failed", Some("missing")); + assert_terminal(&events, "process_telemetry", "failed", Some("missing")); +} + +#[test] +fn invalid_key_uses_a_bounded_reason_without_leaking_the_value() { + let secret = "private-key-material-that-must-not-appear"; + let output = run_relay(&[("BUZZ_RELAY_PRIVATE_KEY", secret)]); + assert!(!output.status.success()); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_terminal(&events, "key_load", "failed", Some("required_invalid")); + let combined = [output.stdout, output.stderr].concat(); + assert!(!String::from_utf8_lossy(&combined).contains(secret)); +} + +#[test] +fn occupied_metrics_port_has_a_typed_bind_terminal() { + let occupied = TcpListener::bind(("0.0.0.0", 0)).expect("bind occupied port"); + let port = occupied.local_addr().expect("occupied address").port(); + let port = port.to_string(); + let output = run_relay(&[ + ("BUZZ_RELAY_PRIVATE_KEY", VALID_RELAY_PRIVATE_KEY), + ("BUZZ_METRICS_PORT", &port), + ]); + assert!(!output.status.success()); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_terminal(&events, "metrics_bind", "failed", Some("bind")); + assert_terminal(&events, "process_telemetry", "failed", Some("bind")); +} + +#[test] +fn otlp_build_failure_is_degraded_without_leaking_endpoint_credentials() { + let secret = "telemetry-secret-marker"; + let endpoint = format!("https://telemetry-user:{secret}@["); + let fake_database = TcpListener::bind(("127.0.0.1", 0)).expect("bind fake database"); + let database_url = format!( + "postgres://buzz@127.0.0.1:{}/buzz", + fake_database.local_addr().expect("database address").port() + ); + let reserved = TcpListener::bind(("127.0.0.1", 0)).expect("reserve metrics port"); + let port = reserved.local_addr().expect("metrics address").port(); + drop(reserved); + let port_value = port.to_string(); + let mut process = RelayProcess::spawn(&[ + ("OTEL_EXPORTER_OTLP_ENDPOINT", &endpoint), + ("RUST_LOG", "buzz_relay=warn"), + ("BUZZ_RELAY_PRIVATE_KEY", VALID_RELAY_PRIVATE_KEY), + ("BUZZ_METRICS_PORT", &port_value), + ("DATABASE_URL", &database_url), + ]); + let scrape = wait_for_relay_metrics(&mut process, port); + assert_no_startup_lifecycle_metrics(&scrape); + let output = process.terminate(); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_terminal(&events, "tracing_init", "degraded", Some("exporter_build")); + assert_terminal( + &events, + "process_telemetry", + "degraded", + Some("exporter_build"), + ); + let combined = [output.stdout, output.stderr].concat(); + assert!(!String::from_utf8_lossy(&combined).contains(secret)); +} + +#[test] +fn successful_main_emits_complete_lifecycle_without_startup_metrics() { + let fake_database = TcpListener::bind(("127.0.0.1", 0)).expect("bind fake database"); + let database_url = format!( + "postgres://buzz@127.0.0.1:{}/buzz", + fake_database.local_addr().expect("database address").port() + ); + let reserved = TcpListener::bind(("127.0.0.1", 0)).expect("reserve metrics port"); + let port = reserved.local_addr().expect("metrics address").port(); + drop(reserved); + let port_value = port.to_string(); + let mut process = RelayProcess::spawn(&[ + ("RUST_LOG", "off"), + ("BUZZ_RELAY_PRIVATE_KEY", VALID_RELAY_PRIVATE_KEY), + ("BUZZ_METRICS_PORT", &port_value), + ("DATABASE_URL", &database_url), + ]); + let scrape = wait_for_relay_metrics(&mut process, port); + assert_no_startup_lifecycle_metrics(&scrape); + + let output = process.terminate(); + let events = lifecycle_events(&output); + assert_accounting(&events); + assert_terminal(&events, "crypto_init", "succeeded", None); + assert_terminal(&events, "tracing_init", "succeeded", None); + assert_terminal(&events, "config_load", "succeeded", None); + assert_terminal(&events, "key_load", "succeeded", None); + assert_terminal(&events, "metrics_bind", "succeeded", None); + assert_terminal(&events, "process_telemetry", "succeeded", None); +} diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index a4545827bca..5e778279130 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -115,6 +115,16 @@ disables that probe through `relay.extraEnv`, `/_readiness` does not test object storage; configuration is still parsed strictly, but reachability and addressing errors surface on the first storage operation. +### Early-startup telemetry contract + +`buzz_process_lifecycle` JSON records are the authoritative history for the +fixed phases `crypto_init`, `tracing_init`, `config_load`, `key_load`, and +`metrics_bind`, plus the aggregate `process_telemetry` result. They use bounded +status/reason values and never contain raw configuration, keys, URLs, or errors. +These phases intentionally do not emit metrics. Most run before the Prometheus +exporter exists, and one uniform log-only contract preserves every phase's real +event time and failure without assigning an eventual scrape time to earlier work. + ### Readiness telemetry contract Only requests served by the private health listener (`BUZZ_HEALTH_PORT`) emit