diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index e096960ad44..30c46e2fd96 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -31,7 +31,8 @@ pub mod error; mod test_support; pub use runtime::{ - insert_mentions, migration, replica_fence, Db, DbConfig, DbPoolStats, ReadSession, + insert_mentions, migration, replica_fence, Db, DbConfig, DbPoolStats, DbReadinessOutcome, + ReadSession, }; pub(crate) use runtime::{ insert_mentions_in_transaction, observability, route_proof, ReadSessionInner, RouteDecision, diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 693d75d66b2..214cc4bca60 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -425,6 +425,26 @@ pub struct DbPoolStats { pub max: u32, } +/// Bounded outcome of the Postgres portion of a relay readiness check. +/// +/// The variants deliberately separate waiting for a pooled connection from +/// executing the health query. Callers may safely use the variant names as +/// low-cardinality metric labels; detailed SQLx errors remain in logs rather +/// than becoming labels. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DbReadinessOutcome { + /// A writer-pool connection was acquired and `SELECT 1` succeeded. + Success, + /// No writer-pool connection became available before the readiness deadline. + PoolTimeout, + /// The writer pool returned a non-timeout acquisition error. + PoolError, + /// A connection was acquired, but `SELECT 1` exceeded the readiness deadline. + QueryTimeout, + /// A connection was acquired, but `SELECT 1` returned an error. + QueryError, +} + /// Configuration for the Postgres connection pool. #[derive(Debug, Clone)] pub struct DbConfig { @@ -931,11 +951,50 @@ impl Db { migration::run_migrations(&self.pool).await } - /// Returns `true` if the database is reachable (used by readiness probes). + /// Returns `true` if the database is reachable. pub async fn ping(&self) -> bool { sqlx::query("SELECT 1").execute(&self.pool).await.is_ok() } + /// Checks writer-pool acquisition and query execution against one deadline. + /// + /// Unlike [`Self::ping`], this preserves whether readiness was blocked while + /// borrowing a connection or failed after a connection had been acquired. + /// The query runs on the already-acquired connection so the two phases + /// cannot be collapsed into a second implicit pool acquisition. + pub async fn readiness_check(&self, deadline: tokio::time::Instant) -> DbReadinessOutcome { + self.readiness_check_sql(deadline, "SELECT 1").await + } + + /// Production-bound seam for classifying failures after pool acquisition. + /// Tests vary only the SQL so timeout/error/cancellation paths execute the + /// same acquisition and classification code as [`Self::readiness_check`]. + async fn readiness_check_sql( + &self, + deadline: tokio::time::Instant, + query: &'static str, + ) -> DbReadinessOutcome { + let mut connection = match tokio::time::timeout_at(deadline, self.pool.acquire()).await { + Err(_) => return DbReadinessOutcome::PoolTimeout, + Ok(Err(sqlx::Error::PoolTimedOut)) => return DbReadinessOutcome::PoolTimeout, + Ok(Err(error)) => { + tracing::debug!(error = %error, "Postgres readiness pool acquisition failed"); + return DbReadinessOutcome::PoolError; + } + Ok(Ok(connection)) => connection, + }; + + match tokio::time::timeout_at(deadline, sqlx::query(query).execute(&mut *connection)).await + { + Err(_) => DbReadinessOutcome::QueryTimeout, + Ok(Err(error)) => { + tracing::debug!(error = %error, "Postgres readiness query failed"); + DbReadinessOutcome::QueryError + } + Ok(Ok(_)) => DbReadinessOutcome::Success, + } + } + /// Returns pool utilisation stats for metrics emission. /// /// `size` — total connections (idle + active) diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index a2938112bc5..3022a15e969 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -19,6 +19,153 @@ async fn setup_db() -> Db { Db::from_pool(pool) } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn readiness_check_distinguishes_pool_exhaustion_from_success() { + let database_url = crate::test_support::database_url(); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&database_url) + .await + .expect("connect size-one readiness test pool"); + let held = pool + .acquire() + .await + .expect("hold the only readiness test connection"); + let db = Db::from_pool(pool); + + let exhausted = db + .readiness_check(tokio::time::Instant::now() + std::time::Duration::from_millis(25)) + .await; + assert_eq!(exhausted, DbReadinessOutcome::PoolTimeout); + + drop(held); + let recovered = db + .readiness_check(tokio::time::Instant::now() + std::time::Duration::from_secs(1)) + .await; + assert_eq!(recovered, DbReadinessOutcome::Success); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn readiness_check_classifies_closed_pool_query_timeout_and_query_error() { + let database_url = crate::test_support::database_url(); + + let closed_pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&database_url) + .await + .expect("connect closed readiness test pool"); + closed_pool.close().await; + let closed = Db::from_pool(closed_pool) + .readiness_check(tokio::time::Instant::now() + std::time::Duration::from_secs(1)) + .await; + assert_eq!(closed, DbReadinessOutcome::PoolError); + + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&database_url) + .await + .expect("connect query classification test pool"); + let db = Db::from_pool(pool); + + let timed_out = db + .readiness_check_sql( + tokio::time::Instant::now() + std::time::Duration::from_millis(25), + "SELECT pg_sleep(0.2)", + ) + .await; + assert_eq!(timed_out, DbReadinessOutcome::QueryTimeout); + + let query_error = db + .readiness_check_sql( + tokio::time::Instant::now() + std::time::Duration::from_secs(1), + "SELECT 1 / 0", + ) + .await; + assert_eq!(query_error, DbReadinessOutcome::QueryError); + + assert_eq!( + db.readiness_check(tokio::time::Instant::now() + std::time::Duration::from_secs(1)) + .await, + DbReadinessOutcome::Success, + "query failures must return the acquired connection to the pool" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn readiness_check_cancellation_balances_waiter_and_inflight_connection() { + let database_url = crate::test_support::database_url(); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&database_url) + .await + .expect("connect cancellation readiness test pool"); + let held = pool + .acquire() + .await + .expect("hold sole connection before waiter cancellation"); + let db = Db::from_pool(pool); + + let waiting_db = db.clone(); + let waiting = tokio::spawn(async move { + waiting_db + .readiness_check(tokio::time::Instant::now() + std::time::Duration::from_secs(5)) + .await + }); + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + waiting.abort(); + assert!(waiting + .await + .expect_err("waiting check must be cancelled") + .is_cancelled()); + drop(held); + + assert_eq!( + db.readiness_check(tokio::time::Instant::now() + std::time::Duration::from_secs(1)) + .await, + DbReadinessOutcome::Success, + "cancelled pool waiter must not consume the released connection" + ); + + let querying_db = db.clone(); + let querying = tokio::spawn(async move { + querying_db + .readiness_check_sql( + tokio::time::Instant::now() + std::time::Duration::from_secs(5), + "SELECT pg_sleep(5)", + ) + .await + }); + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + querying.abort(); + assert!(querying + .await + .expect_err("querying check must be cancelled") + .is_cancelled()); + + let recovered = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let outcome = db + .readiness_check( + tokio::time::Instant::now() + std::time::Duration::from_millis(250), + ) + .await; + match outcome { + DbReadinessOutcome::Success => break outcome, + DbReadinessOutcome::PoolTimeout => tokio::task::yield_now().await, + unexpected => panic!( + "cancelled in-flight query produced unexpected recovery outcome: {unexpected:?}" + ), + } + } + }) + .await + .expect("cancelled in-flight query must return or replace its connection"); + assert_eq!(recovered, DbReadinessOutcome::Success); +} + async fn make_community(pool: &PgPool) -> Uuid { let id = Uuid::new_v4(); let host = format!("communities-of-channels-{}.example", id.simple()); diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index e762c14b1e7..123440c0416 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -35,6 +35,7 @@ pub mod nip11; pub mod protocol; /// Durable NIP-PL matcher and delivery worker. pub mod push_runtime; +mod readiness; /// Axum router construction. pub mod router; /// Shared application state. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index bb8715508e7..260dfaed68b 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1,5 +1,4 @@ use std::collections::{HashMap, HashSet}; -use std::sync::atomic::Ordering; use std::sync::Arc; use tracing::{error, info, warn}; @@ -1312,7 +1311,7 @@ async fn serve( }); let (shutdown_tx, _) = tokio::sync::watch::channel(false); - let shutdown_flag = Arc::clone(&state.shutting_down); + let shutdown_state = Arc::clone(&state); let drain_conn_manager = Arc::clone(&state.conn_manager); let drain_jitter_ms = state.config.drain_jitter_ms; let tx = shutdown_tx.clone(); @@ -1345,7 +1344,7 @@ async fn serve( // sleeps. Not implemented here. This comment records the plan only. let shutdown_handle = tokio::spawn(async move { shutdown_signal().await; - shutdown_flag.store(true, Ordering::Relaxed); + shutdown_state.begin_shutdown(); info!("Shutdown signal received — readiness now returns 503"); // 5s grace: let K8s stop routing new traffic before we close listeners. tokio::time::sleep(std::time::Duration::from_secs(5)).await; diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 16e521a44ee..fb484b01742 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -32,6 +32,11 @@ const LATENCY_BUCKETS_MS: [f64; 11] = [ /// Seconds-scale buckets for internal processing histograms (event, search, audit). const DURATION_BUCKETS_S: [f64; 10] = [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0]; +/// Readiness buckets concentrate resolution near the two-second failure budget. +const READINESS_DURATION_BUCKETS_S: [f64; 15] = [ + 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 0.75, 1.0, 1.25, 1.5, 1.75, 2.0, 2.5, +]; + /// Seconds-scale buckets for Git hydration and pack streams. const GIT_DURATION_BUCKETS_S: [f64; 13] = [ 0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, @@ -56,16 +61,8 @@ const GIT_PACK_BUCKETS: [f64; 9] = [0.0, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 1 /// Integer-count buckets for fan-out recipient histograms. const FANOUT_BUCKETS: [f64; 9] = [0.0, 1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0, 1000.0]; -/// 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) { - let (recorder, exporter) = PrometheusBuilder::new() - .with_http_listener(([0, 0, 0, 0], port)) +fn configured_prometheus_builder(gauge_idle_timeout_secs: u64) -> PrometheusBuilder { + PrometheusBuilder::new() // Remove gauge series that the relay intentionally stops emitting. .idle_timeout( MetricKindMask::GAUGE, @@ -102,6 +99,11 @@ pub fn install(port: u16, gauge_idle_timeout_secs: u64) { &GIT_DURATION_BUCKETS_S, ) .expect("valid git compaction duration bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_readiness_check_duration_seconds".to_owned()), + &READINESS_DURATION_BUCKETS_S, + ) + .expect("valid readiness duration bucket boundaries") .set_buckets_for_metric( Matcher::Full("buzz_git_hydrate_bytes".to_owned()), &GIT_BYTES_BUCKETS, @@ -139,13 +141,57 @@ pub fn install(port: u16, gauge_idle_timeout_secs: u64) { &FANOUT_BUCKETS, ) .expect("valid fanout bucket boundaries") +} + +/// 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) { + 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"); metrics::set_global_recorder(recorder).expect("global recorder must be set exactly once"); + describe_readiness_metrics(); tokio::spawn(exporter); } +/// Register the frozen readiness metric descriptions with the active recorder. +pub(crate) fn describe_readiness_metrics() { + metrics::describe_counter!( + "buzz_readiness_checks_total", + "Kubernetes health-listener readiness probes by terminal bounded reason" + ); + metrics::describe_counter!( + "buzz_readiness_dependency_checks_total", + "Completed readiness dependency attempts by dependency and bounded outcome" + ); + metrics::describe_histogram!( + "buzz_readiness_check_duration_seconds", + metrics::Unit::Seconds, + "Completed readiness check duration without outcome label multiplication" + ); + metrics::describe_gauge!( + "buzz_readiness_state", + "Latest publishable readiness state by check, where 1 is ready and 0 is not ready" + ); +} + +#[cfg(test)] +pub(crate) fn readiness_test_recorder() -> ( + metrics_exporter_prometheus::PrometheusRecorder, + metrics_exporter_prometheus::PrometheusHandle, +) { + let recorder = configured_prometheus_builder(300).build_recorder(); + let handle = recorder.handle(); + (recorder, handle) +} + /// Axum middleware that records CAKE framework HTTP metrics. /// /// Emits: diff --git a/crates/buzz-relay/src/readiness.rs b/crates/buzz-relay/src/readiness.rs new file mode 100644 index 00000000000..36a36c228b2 --- /dev/null +++ b/crates/buzz-relay/src/readiness.rs @@ -0,0 +1,836 @@ +//! Readiness dependency evaluation and ordered metrics publication. +//! +//! [`ReadinessCoordinator`] is process-owned. Its mutex is the linearization +//! point shared by health-probe commits and terminal shutdown, so an older +//! evaluation can never overwrite newer gauges or publish ready after shutdown. + +use std::future::Future; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; +use std::time::Duration; + +use buzz_db::{Db, DbReadinessOutcome}; +use tokio::time::Instant; + +const READINESS_TIMEOUT: Duration = Duration::from_secs(2); + +/// Closed label set exported by `buzz_readiness_checks_total{reason}`. +#[cfg(test)] +pub(crate) const READINESS_REASON_LABELS: [&str; 12] = [ + "ready", + "shutting_down", + "postgres_pool_timeout", + "postgres_pool_error", + "postgres_query_timeout", + "postgres_query_error", + "redis_pool_timeout", + "redis_pool_error", + "deletion_catalog_timeout", + "deletion_catalog_error", + "overall_timeout", + "multiple_dependencies_failed", +]; + +/// Maximum raw Prometheus series emitted by readiness for one pod. +/// +/// - 12 overall reasons +/// - 11 valid dependency/outcome pairs (Postgres 5, Redis 3, catalog 3) +/// - 4 histograms x (15 configured buckets + `+Inf` + count + sum) = 72 +/// - 4 current-state gauges +#[cfg(test)] +pub(crate) const READINESS_RAW_SERIES_PER_POD: usize = 12 + 11 + (4 * 18) + 4; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PostgresOutcome { + Success, + PoolTimeout, + PoolError, + QueryTimeout, + QueryError, +} + +impl PostgresOutcome { + fn label(self) -> &'static str { + match self { + Self::Success => "success", + Self::PoolTimeout => "pool_timeout", + Self::PoolError => "pool_error", + Self::QueryTimeout => "operation_timeout", + Self::QueryError => "operation_error", + } + } + + fn is_success(self) -> bool { + self == Self::Success + } + + fn is_timeout(self) -> bool { + matches!(self, Self::PoolTimeout | Self::QueryTimeout) + } +} + +impl From for PostgresOutcome { + fn from(outcome: DbReadinessOutcome) -> Self { + match outcome { + DbReadinessOutcome::Success => Self::Success, + DbReadinessOutcome::PoolTimeout => Self::PoolTimeout, + DbReadinessOutcome::PoolError => Self::PoolError, + DbReadinessOutcome::QueryTimeout => Self::QueryTimeout, + DbReadinessOutcome::QueryError => Self::QueryError, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RedisOutcome { + Success, + PoolTimeout, + PoolError, +} + +impl RedisOutcome { + fn label(self) -> &'static str { + match self { + Self::Success => "success", + Self::PoolTimeout => "pool_timeout", + Self::PoolError => "pool_error", + } + } + + fn is_success(self) -> bool { + self == Self::Success + } + + fn is_timeout(self) -> bool { + self == Self::PoolTimeout + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DeletionCatalogOutcome { + Success, + OperationTimeout, + OperationError, +} + +impl DeletionCatalogOutcome { + fn label(self) -> &'static str { + match self { + Self::Success => "success", + Self::OperationTimeout => "operation_timeout", + Self::OperationError => "operation_error", + } + } + + fn is_success(self) -> bool { + self == Self::Success + } + + fn is_timeout(self) -> bool { + self == Self::OperationTimeout + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ReadinessReason { + Ready, + ShuttingDown, + PostgresPoolTimeout, + PostgresPoolError, + PostgresQueryTimeout, + PostgresQueryError, + RedisPoolTimeout, + RedisPoolError, + DeletionCatalogTimeout, + DeletionCatalogError, + OverallTimeout, + MultipleDependenciesFailed, +} + +impl ReadinessReason { + pub(crate) fn label(self) -> &'static str { + match self { + Self::Ready => "ready", + Self::ShuttingDown => "shutting_down", + Self::PostgresPoolTimeout => "postgres_pool_timeout", + Self::PostgresPoolError => "postgres_pool_error", + Self::PostgresQueryTimeout => "postgres_query_timeout", + Self::PostgresQueryError => "postgres_query_error", + Self::RedisPoolTimeout => "redis_pool_timeout", + Self::RedisPoolError => "redis_pool_error", + Self::DeletionCatalogTimeout => "deletion_catalog_timeout", + Self::DeletionCatalogError => "deletion_catalog_error", + Self::OverallTimeout => "overall_timeout", + Self::MultipleDependenciesFailed => "multiple_dependencies_failed", + } + } +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct TimedOutcome { + outcome: O, + duration: Duration, +} + +impl TimedOutcome { + #[cfg(test)] + pub(crate) fn new(outcome: O, duration: Duration) -> Self { + Self { outcome, duration } + } +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct ReadinessEvaluation { + postgres: Option>, + redis: Option>, + deletion_catalog: Option>, + pub(crate) reason: ReadinessReason, + total_duration: Duration, +} + +impl ReadinessEvaluation { + pub(crate) fn shutting_down() -> Self { + Self { + postgres: None, + redis: None, + deletion_catalog: None, + reason: ReadinessReason::ShuttingDown, + total_duration: Duration::ZERO, + } + } + + #[cfg(test)] + pub(crate) fn from_results( + postgres: TimedOutcome, + redis: TimedOutcome, + deletion_catalog: TimedOutcome, + total_duration: Duration, + ) -> Self { + Self::for_dependencies(postgres, redis, deletion_catalog, total_duration) + } + + fn for_dependencies( + postgres: TimedOutcome, + redis: TimedOutcome, + deletion_catalog: TimedOutcome, + total_duration: Duration, + ) -> Self { + let reason = final_reason(postgres.outcome, redis.outcome, deletion_catalog.outcome); + Self { + postgres: Some(postgres), + redis: Some(redis), + deletion_catalog: Some(deletion_catalog), + reason, + total_duration, + } + } + + pub(crate) fn is_ready(self) -> bool { + self.reason == ReadinessReason::Ready + } + + pub(crate) fn postgres_ready(self) -> bool { + self.postgres + .is_some_and(|result| result.outcome.is_success()) + } + + pub(crate) fn redis_ready(self) -> bool { + self.redis.is_some_and(|result| result.outcome.is_success()) + } + + pub(crate) fn deletion_catalog_ready(self) -> bool { + self.deletion_catalog + .is_some_and(|result| result.outcome.is_success()) + } + + fn dependencies_ran(self) -> bool { + self.postgres.is_some() || self.redis.is_some() || self.deletion_catalog.is_some() + } +} + +fn final_reason( + postgres: PostgresOutcome, + redis: RedisOutcome, + deletion_catalog: DeletionCatalogOutcome, +) -> ReadinessReason { + let failure_count = usize::from(!postgres.is_success()) + + usize::from(!redis.is_success()) + + usize::from(!deletion_catalog.is_success()); + + if failure_count == 0 { + return ReadinessReason::Ready; + } + if failure_count > 1 { + let all_failures_are_timeouts = (postgres.is_success() || postgres.is_timeout()) + && (redis.is_success() || redis.is_timeout()) + && (deletion_catalog.is_success() || deletion_catalog.is_timeout()); + return if all_failures_are_timeouts { + ReadinessReason::OverallTimeout + } else { + ReadinessReason::MultipleDependenciesFailed + }; + } + + match postgres { + PostgresOutcome::PoolTimeout => ReadinessReason::PostgresPoolTimeout, + PostgresOutcome::PoolError => ReadinessReason::PostgresPoolError, + PostgresOutcome::QueryTimeout => ReadinessReason::PostgresQueryTimeout, + PostgresOutcome::QueryError => ReadinessReason::PostgresQueryError, + PostgresOutcome::Success => match redis { + RedisOutcome::PoolTimeout => ReadinessReason::RedisPoolTimeout, + RedisOutcome::PoolError => ReadinessReason::RedisPoolError, + RedisOutcome::Success => match deletion_catalog { + DeletionCatalogOutcome::OperationTimeout => ReadinessReason::DeletionCatalogTimeout, + DeletionCatalogOutcome::OperationError => ReadinessReason::DeletionCatalogError, + DeletionCatalogOutcome::Success => ReadinessReason::Ready, + }, + }, + } +} + +async fn timed(future: F) -> TimedOutcome +where + F: Future, +{ + let started_at = Instant::now(); + let outcome = future.await; + TimedOutcome { + outcome, + duration: started_at.elapsed(), + } +} + +async fn evaluate_dependencies( + postgres: P, + redis: R, + deletion_catalog: D, +) -> ReadinessEvaluation +where + P: Future, + R: Future, + D: Future, +{ + let started_at = Instant::now(); + let (postgres, redis, deletion_catalog) = + tokio::join!(timed(postgres), timed(redis), timed(deletion_catalog),); + ReadinessEvaluation::for_dependencies(postgres, redis, deletion_catalog, started_at.elapsed()) +} + +async fn redis_check(pool: &deadpool_redis::Pool, deadline: Instant) -> RedisOutcome { + match tokio::time::timeout_at(deadline, pool.get()).await { + Err(_) => RedisOutcome::PoolTimeout, + Ok(Err(error)) => { + tracing::debug!(error = %error, "Redis readiness pool acquisition failed"); + RedisOutcome::PoolError + } + Ok(Ok(_connection)) => RedisOutcome::Success, + } +} + +async fn deletion_catalog_check(db: &Db, deadline: Instant) -> DeletionCatalogOutcome { + match tokio::time::timeout_at(deadline, db.validate_deletion_serving_catalog()).await { + Err(_) => DeletionCatalogOutcome::OperationTimeout, + Ok(Err(error)) => { + tracing::debug!(error = %error, "Deletion catalog readiness validation failed"); + DeletionCatalogOutcome::OperationError + } + Ok(Ok(())) => DeletionCatalogOutcome::Success, + } +} + +#[async_trait::async_trait] +pub(crate) trait ReadinessEvaluator: Send + Sync { + async fn evaluate(&self, db: &Db, redis_pool: &deadpool_redis::Pool) -> ReadinessEvaluation; +} + +struct ProductionReadinessEvaluator; + +#[async_trait::async_trait] +impl ReadinessEvaluator for ProductionReadinessEvaluator { + async fn evaluate(&self, db: &Db, redis_pool: &deadpool_redis::Pool) -> ReadinessEvaluation { + let deadline = Instant::now() + READINESS_TIMEOUT; + evaluate_dependencies( + async { db.readiness_check(deadline).await.into() }, + redis_check(redis_pool, deadline), + deletion_catalog_check(db, deadline), + ) + .await + } +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct ProbeTicket { + generation: u64, +} + +#[derive(Debug, Clone, Copy)] +pub(crate) enum ProbeStart { + Evaluate(ProbeTicket), + ShuttingDown, +} + +#[derive(Debug, Default)] +struct PublicationState { + next_generation: u64, + latest_published_generation: u64, + shutdown_generation: Option, +} + +/// Serializes readiness result publication with terminal process shutdown. +pub(crate) struct ReadinessCoordinator { + state: Mutex, + evaluator: Arc, +} + +impl Default for ReadinessCoordinator { + fn default() -> Self { + Self { + state: Mutex::new(PublicationState::default()), + evaluator: Arc::new(ProductionReadinessEvaluator), + } + } +} + +impl ReadinessCoordinator { + #[cfg(test)] + pub(crate) fn with_evaluator(evaluator: Arc) -> Self { + Self { + state: Mutex::new(PublicationState::default()), + evaluator, + } + } + + fn lock_state(&self) -> MutexGuard<'_, PublicationState> { + self.state.lock().unwrap_or_else(PoisonError::into_inner) + } + + pub(crate) async fn evaluate( + &self, + db: &Db, + redis_pool: &deadpool_redis::Pool, + ) -> ReadinessEvaluation { + self.evaluator.evaluate(db, redis_pool).await + } + + /// Allocates a health-probe generation or records a truthful shutdown fast path. + pub(crate) fn begin_probe(&self) -> ProbeStart { + let mut state = self.lock_state(); + if state.shutdown_generation.is_some() { + let evaluation = ReadinessEvaluation::shutting_down(); + record_attempt_metrics(&evaluation, ReadinessReason::ShuttingDown); + record_overall_state(false); + return ProbeStart::ShuttingDown; + } + + state.next_generation = state.next_generation.saturating_add(1); + ProbeStart::Evaluate(ProbeTicket { + generation: state.next_generation, + }) + } + + /// Commits one completed health probe through the shared publication fence. + pub(crate) fn finish_probe( + &self, + ticket: ProbeTicket, + evaluation: ReadinessEvaluation, + ) -> ReadinessEvaluation { + let mut state = self.lock_state(); + if state.shutdown_generation.is_some() { + record_attempt_metrics(&evaluation, ReadinessReason::ShuttingDown); + return ReadinessEvaluation::shutting_down(); + } + + record_attempt_metrics(&evaluation, evaluation.reason); + if ticket.generation > state.latest_published_generation { + record_current_state(&evaluation); + state.latest_published_generation = ticket.generation; + } + evaluation + } + + /// Returns whether a compatibility/public readiness evaluation may start. + pub(crate) fn public_evaluation_allowed(&self) -> bool { + self.lock_state().shutdown_generation.is_none() + } + + /// Makes shutdown dominate a public request that was already in flight. + pub(crate) fn finish_public_evaluation( + &self, + evaluation: ReadinessEvaluation, + ) -> ReadinessEvaluation { + if self.lock_state().shutdown_generation.is_some() { + ReadinessEvaluation::shutting_down() + } else { + evaluation + } + } + + /// Commits terminal shutdown and immediately publishes overall not-ready. + pub(crate) fn begin_shutdown(&self) { + let mut state = self.lock_state(); + if state.shutdown_generation.is_none() { + let generation = state.next_generation.saturating_add(1); + state.shutdown_generation = Some(generation); + record_overall_state(false); + } + } +} + +fn record_attempt_metrics(evaluation: &ReadinessEvaluation, reason: ReadinessReason) { + metrics::counter!( + "buzz_readiness_checks_total", + "reason" => reason.label(), + ) + .increment(1); + + if !evaluation.dependencies_ran() { + return; + } + + metrics::histogram!( + "buzz_readiness_check_duration_seconds", + "check" => "overall", + ) + .record(evaluation.total_duration.as_secs_f64()); + + if let Some(result) = evaluation.postgres { + record_dependency_attempt("postgres", result.outcome.label(), result.duration); + } + if let Some(result) = evaluation.redis { + record_dependency_attempt("redis", result.outcome.label(), result.duration); + } + if let Some(result) = evaluation.deletion_catalog { + record_dependency_attempt("deletion_catalog", result.outcome.label(), result.duration); + } +} + +fn record_dependency_attempt(dependency: &'static str, outcome: &'static str, duration: Duration) { + metrics::counter!( + "buzz_readiness_dependency_checks_total", + "dependency" => dependency, + "outcome" => outcome, + ) + .increment(1); + metrics::histogram!( + "buzz_readiness_check_duration_seconds", + "check" => dependency, + ) + .record(duration.as_secs_f64()); +} + +fn record_current_state(evaluation: &ReadinessEvaluation) { + record_overall_state(evaluation.is_ready()); + if let Some(result) = evaluation.postgres { + record_dependency_state("postgres", result.outcome.is_success()); + } + if let Some(result) = evaluation.redis { + record_dependency_state("redis", result.outcome.is_success()); + } + if let Some(result) = evaluation.deletion_catalog { + record_dependency_state("deletion_catalog", result.outcome.is_success()); + } +} + +fn record_overall_state(ready: bool) { + metrics::gauge!("buzz_readiness_state", "check" => "overall").set(if ready { + 1.0 + } else { + 0.0 + }); +} + +fn record_dependency_state(dependency: &'static str, ready: bool) { + metrics::gauge!("buzz_readiness_state", "check" => dependency).set(if ready { + 1.0 + } else { + 0.0 + }); +} + +#[cfg(test)] +mod tests { + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use metrics_util::CompositeKey; + + use super::*; + + fn ready_evaluation() -> ReadinessEvaluation { + ReadinessEvaluation::from_results( + TimedOutcome::new(PostgresOutcome::Success, Duration::from_millis(35)), + TimedOutcome::new(RedisOutcome::Success, Duration::from_millis(10)), + TimedOutcome::new(DeletionCatalogOutcome::Success, Duration::from_millis(20)), + Duration::from_millis(35), + ) + } + + fn redis_failure_evaluation() -> ReadinessEvaluation { + ReadinessEvaluation::from_results( + TimedOutcome::new(PostgresOutcome::Success, Duration::from_millis(35)), + TimedOutcome::new(RedisOutcome::PoolTimeout, Duration::from_secs(2)), + TimedOutcome::new(DeletionCatalogOutcome::Success, Duration::from_millis(20)), + Duration::from_secs(2), + ) + } + + fn exact_metric<'a>( + snapshot: &'a [( + CompositeKey, + Option, + Option, + DebugValue, + )], + name: &str, + labels: &[(&str, &str)], + ) -> Option<&'a DebugValue> { + snapshot.iter().find_map(|(key, _, _, value)| { + let actual = key + .key() + .labels() + .map(|label| (label.key(), label.value())) + .collect::>(); + (key.key().name() == name + && actual.len() == labels.len() + && labels.iter().all(|expected| actual.contains(expected))) + .then_some(value) + }) + } + + fn gauge_value( + snapshot: &[( + CompositeKey, + Option, + Option, + DebugValue, + )], + check: &str, + ) -> f64 { + let value = exact_metric(snapshot, "buzz_readiness_state", &[("check", check)]) + .expect("readiness gauge"); + let DebugValue::Gauge(value) = value else { + panic!("readiness state must be a gauge"); + }; + value.into_inner() + } + + #[tokio::test(start_paused = true)] + async fn evaluation_preserves_a_completed_check_when_another_times_out() { + let evaluation = evaluate_dependencies( + async { + tokio::time::sleep(Duration::from_millis(35)).await; + PostgresOutcome::Success + }, + async { + tokio::time::sleep(Duration::from_secs(2)).await; + RedisOutcome::PoolTimeout + }, + async { + tokio::time::sleep(Duration::from_millis(10)).await; + DeletionCatalogOutcome::Success + }, + ) + .await; + + assert_eq!(evaluation.reason, ReadinessReason::RedisPoolTimeout); + assert_eq!( + evaluation.postgres.map(|result| result.duration), + Some(Duration::from_millis(35)) + ); + assert_eq!( + evaluation.redis.map(|result| result.duration), + Some(Duration::from_secs(2)) + ); + } + + #[test] + fn simultaneous_dependency_timeouts_are_an_overall_timeout() { + assert_eq!( + final_reason( + PostgresOutcome::PoolTimeout, + RedisOutcome::PoolTimeout, + DeletionCatalogOutcome::Success, + ), + ReadinessReason::OverallTimeout + ); + } + + #[test] + fn dependency_types_expose_only_valid_outcome_pairs() { + assert_eq!( + [ + PostgresOutcome::Success, + PostgresOutcome::PoolTimeout, + PostgresOutcome::PoolError, + PostgresOutcome::QueryTimeout, + PostgresOutcome::QueryError, + ] + .map(PostgresOutcome::label), + [ + "success", + "pool_timeout", + "pool_error", + "operation_timeout", + "operation_error", + ] + ); + assert_eq!( + [ + RedisOutcome::Success, + RedisOutcome::PoolTimeout, + RedisOutcome::PoolError, + ] + .map(RedisOutcome::label), + ["success", "pool_timeout", "pool_error"] + ); + assert_eq!( + [ + DeletionCatalogOutcome::Success, + DeletionCatalogOutcome::OperationTimeout, + DeletionCatalogOutcome::OperationError, + ] + .map(DeletionCatalogOutcome::label), + ["success", "operation_timeout", "operation_error"] + ); + assert_eq!(READINESS_RAW_SERIES_PER_POD, 99); + } + + #[test] + fn slow_older_failure_cannot_overwrite_newer_success_gauges() { + let coordinator = ReadinessCoordinator::default(); + let ProbeStart::Evaluate(slow_a) = coordinator.begin_probe() else { + panic!("serving probe A"); + }; + let ProbeStart::Evaluate(fast_b) = coordinator.begin_probe() else { + panic!("serving probe B"); + }; + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + coordinator.finish_probe(fast_b, ready_evaluation()); + coordinator.finish_probe(slow_a, redis_failure_evaluation()); + }); + let snapshot = snapshotter.snapshot().into_vec(); + + assert_eq!(gauge_value(&snapshot, "overall"), 1.0); + assert_eq!(gauge_value(&snapshot, "redis"), 1.0); + assert!(matches!( + exact_metric( + &snapshot, + "buzz_readiness_checks_total", + &[("reason", "ready")] + ), + Some(DebugValue::Counter(1)) + )); + assert!(matches!( + exact_metric( + &snapshot, + "buzz_readiness_checks_total", + &[("reason", "redis_pool_timeout")] + ), + Some(DebugValue::Counter(1)) + )); + } + + #[test] + fn slow_older_success_cannot_overwrite_newer_failure_gauges() { + let coordinator = ReadinessCoordinator::default(); + let ProbeStart::Evaluate(slow_a) = coordinator.begin_probe() else { + panic!("serving probe A"); + }; + let ProbeStart::Evaluate(fast_b) = coordinator.begin_probe() else { + panic!("serving probe B"); + }; + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + coordinator.finish_probe(fast_b, redis_failure_evaluation()); + coordinator.finish_probe(slow_a, ready_evaluation()); + }); + let snapshot = snapshotter.snapshot().into_vec(); + + assert_eq!(gauge_value(&snapshot, "overall"), 0.0); + assert_eq!(gauge_value(&snapshot, "postgres"), 1.0); + assert_eq!(gauge_value(&snapshot, "redis"), 0.0); + assert_eq!(gauge_value(&snapshot, "deletion_catalog"), 1.0); + } + + #[test] + fn shutdown_fast_path_preserves_dependency_state_and_histograms() { + let coordinator = ReadinessCoordinator::default(); + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + metrics::with_local_recorder(&recorder, || { + let ProbeStart::Evaluate(ticket) = coordinator.begin_probe() else { + panic!("initial serving probe"); + }; + coordinator.finish_probe(ticket, ready_evaluation()); + coordinator.begin_shutdown(); + assert!(matches!( + coordinator.begin_probe(), + ProbeStart::ShuttingDown + )); + }); + let after = snapshotter.snapshot().into_vec(); + + for dependency in ["postgres", "redis", "deletion_catalog"] { + assert_eq!( + gauge_value(&after, dependency), + 1.0, + "shutdown must not fabricate {dependency} state" + ); + } + for check in ["overall", "postgres", "redis", "deletion_catalog"] { + assert!( + matches!( + exact_metric( + &after, + "buzz_readiness_check_duration_seconds", + &[("check", check)] + ), + Some(DebugValue::Histogram(values)) if values.len() == 1 + ), + "shutdown fast path must not add a {check} duration" + ); + } + assert_eq!(gauge_value(&after, "overall"), 0.0); + assert!(matches!( + exact_metric( + &after, + "buzz_readiness_checks_total", + &[("reason", "shutting_down")] + ), + Some(DebugValue::Counter(1)) + )); + } + + #[test] + fn shutdown_dominates_an_in_flight_success_without_resurrecting_gauges() { + let coordinator = ReadinessCoordinator::default(); + let ProbeStart::Evaluate(ticket) = coordinator.begin_probe() else { + panic!("serving probe"); + }; + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + let response = metrics::with_local_recorder(&recorder, || { + coordinator.begin_shutdown(); + coordinator.finish_probe(ticket, ready_evaluation()) + }); + let snapshot = snapshotter.snapshot().into_vec(); + + assert_eq!(response.reason, ReadinessReason::ShuttingDown); + assert_eq!(gauge_value(&snapshot, "overall"), 0.0); + assert!( + exact_metric(&snapshot, "buzz_readiness_state", &[("check", "postgres")]).is_none() + ); + assert!(matches!( + exact_metric( + &snapshot, + "buzz_readiness_dependency_checks_total", + &[("dependency", "postgres"), ("outcome", "success")] + ), + Some(DebugValue::Counter(1)) + )); + } +} diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index dd0fde6fdcd..61aedf70be0 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -24,6 +24,7 @@ use crate::audio; use crate::connection::handle_connection; use crate::metrics::track_metrics; use crate::nip11::{nip11_document, relay_info_handler}; +use crate::readiness::{self, ReadinessEvaluation, ReadinessReason}; use crate::state::AppState; /// Build the axum [`Router`] with all relay routes, middleware, and CORS configuration. @@ -67,7 +68,7 @@ pub fn build_router(state: Arc) -> Router { // Health endpoints .route("/health", get(health_handler)) .route("/_liveness", get(liveness_handler)) - .route("/_readiness", get(readiness_handler)) + .route("/_readiness", get(public_readiness_handler)) // Nostr HTTP bridge (NIP-98 auth) .route("/events", post(api::bridge::submit_event)) .route("/query", post(api::bridge::query_events)) @@ -294,7 +295,7 @@ async fn admin_spa_document(state: &AppState, accept: &str) -> axum::response::R pub fn build_health_router(state: Arc) -> Router { Router::new() .route("/_liveness", get(liveness_handler)) - .route("/_readiness", get(readiness_handler)) + .route("/_readiness", get(kubernetes_readiness_handler)) .route("/_status", get(status_handler)) .route("/_mesh", get(mesh_status_handler)) .with_state(state) @@ -406,11 +407,36 @@ async fn liveness_handler() -> impl IntoResponse { (StatusCode::OK, "ok") } -/// Readiness probe — checks shutdown flag, Postgres, and Redis connectivity. -async fn readiness_handler(State(state): State>) -> impl IntoResponse { - use std::time::Duration; +/// Compatibility endpoint on the public listener. It evaluates dependencies +/// and preserves the existing response contract but never records rollout +/// telemetry. +async fn public_readiness_handler(State(state): State>) -> impl IntoResponse { + if !state.readiness.public_evaluation_allowed() { + return readiness_response(ReadinessEvaluation::shutting_down(), false); + } + + let evaluation = state.readiness.evaluate(&state.db, &state.redis_pool).await; + let evaluation = state.readiness.finish_public_evaluation(evaluation); + readiness_response(evaluation, false) +} + +/// Kubernetes health-listener endpoint. All rollout metrics flow through the +/// process-owned coordinator so shutdown and probe generations are ordered. +async fn kubernetes_readiness_handler(State(state): State>) -> impl IntoResponse { + let readiness::ProbeStart::Evaluate(ticket) = state.readiness.begin_probe() else { + return readiness_response(ReadinessEvaluation::shutting_down(), true); + }; + + let evaluation = state.readiness.evaluate(&state.db, &state.redis_pool).await; + let evaluation = state.readiness.finish_probe(ticket, evaluation); + readiness_response(evaluation, true) +} - if state.shutting_down.load(Ordering::Relaxed) { +fn readiness_response( + evaluation: ReadinessEvaluation, + include_reason: bool, +) -> axum::response::Response { + if evaluation.reason == ReadinessReason::ShuttingDown { return ( StatusCode::SERVICE_UNAVAILABLE, Json(json!({"status": "shutting_down"})), @@ -418,33 +444,23 @@ async fn readiness_handler(State(state): State>) -> impl IntoRespo .into_response(); } - let check = async { - let (pg_ok, redis_ok, deletion_catalog_ok) = tokio::join!( - state.db.ping(), - async { state.redis_pool.get().await.is_ok() }, - async { state.db.validate_deletion_serving_catalog().await.is_ok() }, - ); - (pg_ok, redis_ok, deletion_catalog_ok) - }; - - let (pg_ok, redis_ok, deletion_catalog_ok) = - tokio::time::timeout(Duration::from_secs(2), check) - .await - .unwrap_or((false, false, false)); + let pg_ok = evaluation.postgres_ready(); + let redis_ok = evaluation.redis_ready(); + let deletion_catalog_ok = evaluation.deletion_catalog_ready(); - if pg_ok && redis_ok && deletion_catalog_ok { + if evaluation.is_ready() { (StatusCode::OK, Json(json!({"status": "ready"}))).into_response() } else { - ( - StatusCode::SERVICE_UNAVAILABLE, - Json(json!({ - "status": "not_ready", - "postgres": pg_ok, - "redis": redis_ok, - "deletion_catalog": deletion_catalog_ok - })), - ) - .into_response() + let mut payload = json!({ + "status": "not_ready", + "postgres": pg_ok, + "redis": redis_ok, + "deletion_catalog": deletion_catalog_ok + }); + if include_reason { + payload["reason"] = json!(evaluation.reason.label()); + } + (StatusCode::SERVICE_UNAVAILABLE, Json(payload)).into_response() } } @@ -506,12 +522,17 @@ fn build_cors_layer(cors_origins: &[String]) -> CorsLayer { #[cfg(test)] mod tests { + use std::collections::VecDeque; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + use std::sync::{Mutex, PoisonError}; + use std::time::Duration; + use axum::{routing::get, Router}; use futures_util::SinkExt; use opentelemetry::trace::TracerProvider as _; use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider}; use tokio::net::TcpListener; - use tokio::sync::mpsc; + use tokio::sync::{mpsc, Notify}; use tokio_tungstenite::{connect_async, tungstenite::Message}; use tower::ServiceBuilder; use tracing::Instrument as _; @@ -519,6 +540,98 @@ mod tests { use super::*; + struct ScriptedReadinessEvaluator { + evaluations: Mutex>, + } + + impl ScriptedReadinessEvaluator { + fn new(evaluations: impl IntoIterator) -> Self { + Self { + evaluations: Mutex::new(evaluations.into_iter().collect()), + } + } + + fn push(&self, evaluation: ReadinessEvaluation) { + self.evaluations + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push_back(evaluation); + } + } + + #[async_trait::async_trait] + impl readiness::ReadinessEvaluator for ScriptedReadinessEvaluator { + async fn evaluate( + &self, + _db: &buzz_db::Db, + _redis_pool: &deadpool_redis::Pool, + ) -> ReadinessEvaluation { + self.evaluations + .lock() + .unwrap_or_else(PoisonError::into_inner) + .pop_front() + .expect("scripted readiness evaluation") + } + } + + struct BarrierReadinessEvaluator { + calls: AtomicUsize, + first_started: Notify, + release_first: Notify, + first: ReadinessEvaluation, + second: ReadinessEvaluation, + } + + impl BarrierReadinessEvaluator { + fn new(first: ReadinessEvaluation, second: ReadinessEvaluation) -> Self { + Self { + calls: AtomicUsize::new(0), + first_started: Notify::new(), + release_first: Notify::new(), + first, + second, + } + } + } + + #[async_trait::async_trait] + impl readiness::ReadinessEvaluator for BarrierReadinessEvaluator { + async fn evaluate( + &self, + _db: &buzz_db::Db, + _redis_pool: &deadpool_redis::Pool, + ) -> ReadinessEvaluation { + if self.calls.fetch_add(1, AtomicOrdering::SeqCst) == 0 { + self.first_started.notify_waiters(); + self.release_first.notified().await; + self.first + } else { + self.second + } + } + } + + fn readiness_evaluation( + postgres: readiness::PostgresOutcome, + redis: readiness::RedisOutcome, + deletion_catalog: readiness::DeletionCatalogOutcome, + ) -> ReadinessEvaluation { + ReadinessEvaluation::from_results( + readiness::TimedOutcome::new(postgres, Duration::from_millis(35)), + readiness::TimedOutcome::new(redis, Duration::from_millis(20)), + readiness::TimedOutcome::new(deletion_catalog, Duration::from_millis(15)), + Duration::from_millis(35), + ) + } + + fn ready_evaluation() -> ReadinessEvaluation { + readiness_evaluation( + readiness::PostgresOutcome::Success, + readiness::RedisOutcome::Success, + readiness::DeletionCatalogOutcome::Success, + ) + } + #[test] fn invite_landing_path_requires_exactly_one_nonempty_code_segment() { assert!(is_invite_landing_path("/invite/payload.mac")); @@ -594,6 +707,447 @@ mod tests { Arc::new(state) } + async fn readiness_state(evaluator: Arc) -> Arc { + let mut config = crate::config::Config::from_env().expect("default config loads"); + config.require_relay_membership = false; + config.database_url = "postgres://buzz:buzz_dev@127.0.0.1:1/buzz".to_string(); + config.redis_url = "redis://127.0.0.1:1".to_string(); + let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); + let db = buzz_db::Db::from_pool(pool.clone()); + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .expect("pubsub manager"), + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage"); + let (mut state, _audit_shutdown) = AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + nostr::Keys::generate(), + media_storage, + ); + state.set_readiness_evaluator(evaluator); + Arc::new(state) + } + + async fn readiness_request(router: Router) -> (StatusCode, serde_json::Value) { + let response = router + .oneshot( + Request::get("/_readiness") + .body(Body::empty()) + .expect("readiness request"), + ) + .await + .expect("readiness response"); + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), 64 * 1024) + .await + .expect("readiness response body"); + let payload = serde_json::from_slice(&body).expect("readiness JSON"); + (status, payload) + } + + fn readiness_metric_lines(rendered: &str) -> Vec<&str> { + rendered + .lines() + .filter(|line| line.starts_with("buzz_readiness")) + .collect() + } + + fn sorted_readiness_metric_lines(rendered: &str) -> Vec { + let mut lines = readiness_metric_lines(rendered) + .into_iter() + .map(str::to_owned) + .collect::>(); + lines.sort(); + lines + } + + fn metric_value(rendered: &str, exact_prefix: &str) -> f64 { + rendered + .lines() + .find_map(|line| { + line.strip_prefix(exact_prefix) + .and_then(|value| value.strip_prefix(' ')) + .and_then(|value| value.parse().ok()) + }) + .unwrap_or_else(|| panic!("missing metric line: {exact_prefix}")) + } + + #[test] + fn production_readiness_routes_export_the_frozen_health_only_contract() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime"); + let evaluator = Arc::new(ScriptedReadinessEvaluator::new(std::iter::repeat_n( + ready_evaluation(), + 4, + ))); + let (recorder, handle) = crate::metrics::readiness_test_recorder(); + + metrics::with_local_recorder(&recorder, || { + crate::metrics::describe_readiness_metrics(); + runtime.block_on(async { + let state = readiness_state(evaluator.clone()).await; + let public = build_router(state.clone()); + let health = build_health_router(state.clone()); + + for _ in 0..3 { + assert_eq!( + readiness_request(public.clone()).await, + (StatusCode::OK, json!({"status": "ready"})) + ); + } + assert!( + readiness_metric_lines(&handle.render()).is_empty(), + "public compatibility requests must emit no readiness series" + ); + + assert_eq!( + readiness_request(health.clone()).await, + (StatusCode::OK, json!({"status": "ready"})) + ); + let first_scrape = handle.render(); + + assert!(first_scrape.contains("# TYPE buzz_readiness_checks_total counter")); + assert!(first_scrape + .contains("# TYPE buzz_readiness_dependency_checks_total counter")); + assert!(first_scrape + .contains("# TYPE buzz_readiness_check_duration_seconds histogram")); + assert!(first_scrape.contains("# TYPE buzz_readiness_state gauge")); + assert_eq!( + metric_value( + &first_scrape, + "buzz_readiness_checks_total{reason=\"ready\"}" + ), + 1.0 + ); + assert_eq!( + metric_value( + &first_scrape, + "buzz_readiness_dependency_checks_total{dependency=\"postgres\",outcome=\"success\"}" + ), + 1.0 + ); + assert_eq!( + metric_value( + &first_scrape, + "buzz_readiness_state{check=\"overall\"}" + ), + 1.0 + ); + for bucket in ["2", "2.5", "+Inf"] { + assert!(first_scrape.contains(&format!( + "buzz_readiness_check_duration_seconds_bucket{{check=\"overall\",le=\"{bucket}\"}}" + ))); + } + assert!(!first_scrape.contains("result=")); + assert!(!first_scrape + .lines() + .filter(|line| line.starts_with("buzz_readiness_check_duration_seconds")) + .any(|line| line.contains("outcome="))); + + let before_public_failure = sorted_readiness_metric_lines(&first_scrape); + evaluator.push(readiness_evaluation( + readiness::PostgresOutcome::Success, + readiness::RedisOutcome::PoolTimeout, + readiness::DeletionCatalogOutcome::Success, + )); + assert_eq!( + readiness_request(public.clone()).await, + ( + StatusCode::SERVICE_UNAVAILABLE, + json!({ + "status": "not_ready", + "postgres": true, + "redis": false, + "deletion_catalog": true + }) + ) + ); + assert_eq!( + sorted_readiness_metric_lines(&handle.render()), + before_public_failure + ); + + let contract_evaluations = [ + readiness_evaluation( + readiness::PostgresOutcome::PoolTimeout, + readiness::RedisOutcome::Success, + readiness::DeletionCatalogOutcome::Success, + ), + readiness_evaluation( + readiness::PostgresOutcome::PoolError, + readiness::RedisOutcome::Success, + readiness::DeletionCatalogOutcome::Success, + ), + readiness_evaluation( + readiness::PostgresOutcome::QueryTimeout, + readiness::RedisOutcome::Success, + readiness::DeletionCatalogOutcome::Success, + ), + readiness_evaluation( + readiness::PostgresOutcome::QueryError, + readiness::RedisOutcome::Success, + readiness::DeletionCatalogOutcome::Success, + ), + readiness_evaluation( + readiness::PostgresOutcome::Success, + readiness::RedisOutcome::PoolTimeout, + readiness::DeletionCatalogOutcome::Success, + ), + readiness_evaluation( + readiness::PostgresOutcome::Success, + readiness::RedisOutcome::PoolError, + readiness::DeletionCatalogOutcome::Success, + ), + readiness_evaluation( + readiness::PostgresOutcome::Success, + readiness::RedisOutcome::Success, + readiness::DeletionCatalogOutcome::OperationTimeout, + ), + readiness_evaluation( + readiness::PostgresOutcome::Success, + readiness::RedisOutcome::Success, + readiness::DeletionCatalogOutcome::OperationError, + ), + readiness_evaluation( + readiness::PostgresOutcome::PoolTimeout, + readiness::RedisOutcome::PoolTimeout, + readiness::DeletionCatalogOutcome::OperationTimeout, + ), + readiness_evaluation( + readiness::PostgresOutcome::PoolError, + readiness::RedisOutcome::PoolError, + readiness::DeletionCatalogOutcome::Success, + ), + ]; + for evaluation in contract_evaluations { + evaluator.push(evaluation); + let (status, payload) = readiness_request(health.clone()).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(payload["reason"], json!(evaluation.reason.label())); + } + + let before_shutdown = handle.render(); + let histogram_counts_before = ["overall", "postgres", "redis", "deletion_catalog"] + .map(|check| { + metric_value( + &before_shutdown, + &format!( + "buzz_readiness_check_duration_seconds_count{{check=\"{check}\"}}" + ), + ) + }); + state.begin_shutdown(); + assert_eq!( + readiness_request(public).await, + ( + StatusCode::SERVICE_UNAVAILABLE, + json!({"status": "shutting_down"}) + ) + ); + let after_public_shutdown = handle.render(); + assert!(after_public_shutdown + .lines() + .all(|line| !line.contains("reason=\"shutting_down\""))); + + assert_eq!( + readiness_request(health).await, + ( + StatusCode::SERVICE_UNAVAILABLE, + json!({"status": "shutting_down"}) + ) + ); + let final_scrape = handle.render(); + let histogram_counts_after = ["overall", "postgres", "redis", "deletion_catalog"] + .map(|check| { + metric_value( + &final_scrape, + &format!( + "buzz_readiness_check_duration_seconds_count{{check=\"{check}\"}}" + ), + ) + }); + assert_eq!(histogram_counts_after, histogram_counts_before); + assert_eq!( + metric_value( + &final_scrape, + "buzz_readiness_checks_total{reason=\"shutting_down\"}" + ), + 1.0 + ); + assert_eq!( + metric_value( + &final_scrape, + "buzz_readiness_state{check=\"overall\"}" + ), + 0.0 + ); + assert!(!final_scrape.contains("sensitive-sql-or-url")); + + let exported_reasons = final_scrape + .lines() + .filter(|line| line.starts_with("buzz_readiness_checks_total{")) + .count(); + assert_eq!(exported_reasons, readiness::READINESS_REASON_LABELS.len()); + assert_eq!( + readiness_metric_lines(&final_scrape).len(), + readiness::READINESS_RAW_SERIES_PER_POD, + "readiness series contract must stay at or below its 99-series cap" + ); + }); + }); + } + + fn run_out_of_order_route_case( + first: ReadinessEvaluation, + second: ReadinessEvaluation, + ) -> (serde_json::Value, serde_json::Value, String) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime"); + let evaluator = Arc::new(BarrierReadinessEvaluator::new(first, second)); + let (recorder, handle) = crate::metrics::readiness_test_recorder(); + + metrics::with_local_recorder(&recorder, || { + runtime.block_on(async { + let state = readiness_state(evaluator.clone()).await; + let health = build_health_router(state); + let first_started = evaluator.first_started.notified(); + let slow_first = tokio::spawn(readiness_request(health.clone())); + first_started.await; + + let (_, second_payload) = readiness_request(health).await; + evaluator.release_first.notify_one(); + let (_, first_payload) = slow_first.await.expect("slow first probe task"); + (first_payload, second_payload, handle.render()) + }) + }) + } + + #[test] + fn real_health_route_generation_fence_covers_both_completion_orders() { + let failure = readiness_evaluation( + readiness::PostgresOutcome::Success, + readiness::RedisOutcome::PoolTimeout, + readiness::DeletionCatalogOutcome::Success, + ); + + let (older_failure, newer_success, success_scrape) = + run_out_of_order_route_case(failure, ready_evaluation()); + assert_eq!(older_failure["reason"], json!("redis_pool_timeout")); + assert_eq!(newer_success, json!({"status": "ready"})); + assert_eq!( + metric_value(&success_scrape, "buzz_readiness_state{check=\"overall\"}"), + 1.0 + ); + assert_eq!( + metric_value(&success_scrape, "buzz_readiness_state{check=\"redis\"}"), + 1.0 + ); + + let (older_success, newer_failure, failure_scrape) = + run_out_of_order_route_case(ready_evaluation(), failure); + assert_eq!(older_success, json!({"status": "ready"})); + assert_eq!(newer_failure["reason"], json!("redis_pool_timeout")); + assert_eq!( + metric_value(&failure_scrape, "buzz_readiness_state{check=\"overall\"}"), + 0.0 + ); + assert_eq!( + metric_value(&failure_scrape, "buzz_readiness_state{check=\"redis\"}"), + 0.0 + ); + for scrape in [&success_scrape, &failure_scrape] { + assert_eq!( + metric_value(scrape, "buzz_readiness_checks_total{reason=\"ready\"}"), + 1.0 + ); + assert_eq!( + metric_value( + scrape, + "buzz_readiness_checks_total{reason=\"redis_pool_timeout\"}" + ), + 1.0 + ); + } + } + + #[test] + fn real_health_route_shutdown_fence_dominates_an_in_flight_success() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime"); + let evaluator = Arc::new(BarrierReadinessEvaluator::new( + ready_evaluation(), + ready_evaluation(), + )); + let (recorder, handle) = crate::metrics::readiness_test_recorder(); + + metrics::with_local_recorder(&recorder, || { + runtime.block_on(async { + let state = readiness_state(evaluator.clone()).await; + let health = build_health_router(state.clone()); + let first_started = evaluator.first_started.notified(); + let in_flight = tokio::spawn(readiness_request(health)); + first_started.await; + + state.begin_shutdown(); + evaluator.release_first.notify_one(); + assert_eq!( + in_flight.await.expect("in-flight readiness task"), + ( + StatusCode::SERVICE_UNAVAILABLE, + json!({"status": "shutting_down"}) + ) + ); + + let scrape = handle.render(); + assert_eq!( + metric_value(&scrape, "buzz_readiness_state{check=\"overall\"}"), + 0.0 + ); + assert!(scrape + .lines() + .all(|line| !line.starts_with("buzz_readiness_state{check=\"postgres\"}"))); + assert_eq!( + metric_value( + &scrape, + "buzz_readiness_checks_total{reason=\"shutting_down\"}" + ), + 1.0 + ); + assert_eq!( + metric_value( + &scrape, + "buzz_readiness_dependency_checks_total{dependency=\"postgres\",outcome=\"success\"}" + ), + 1.0 + ); + }); + }); + } + /// A minimal built SPA: an index document, one hashed asset, and the /// root-level favicon Vite copies out of `public/`. fn write_bundle(dir: &std::path::Path) { diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 693f6c7a9bc..47665db6779 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -713,6 +713,8 @@ pub struct AppState { pub audio_rooms: Arc, /// Set to `true` on SIGTERM — readiness probe returns 503. pub shutting_down: Arc, + /// Orders readiness gauge publication against terminal shutdown. + pub(crate) readiness: Arc, /// Process start time — used by `/_status` endpoint. pub started_at: Instant, /// Shared, community-scoped NIP-98 replay prevention. @@ -914,6 +916,7 @@ impl AppState { git_pack_cache, audio_rooms: Arc::new(AudioRoomManager::new()), shutting_down: Arc::new(AtomicBool::new(false)), + readiness: Arc::new(crate::readiness::ReadinessCoordinator::default()), started_at: Instant::now(), nip98_replay, gif_http_client, @@ -955,6 +958,23 @@ impl AppState { ) } + /// Atomically closes readiness publication before exposing shutdown to + /// the relay's other fast-path lifecycle checks. + pub fn begin_shutdown(&self) { + self.readiness.begin_shutdown(); + self.shutting_down.store(true, Ordering::Release); + } + + #[cfg(test)] + pub(crate) fn set_readiness_evaluator( + &mut self, + evaluator: Arc, + ) { + self.readiness = Arc::new(crate::readiness::ReadinessCoordinator::with_evaluator( + evaluator, + )); + } + /// Inter-relay mesh handle. `None` ⇒ mesh-off / single-instance: callers /// must no-op to today's behavior. Set once by `main.rs` after boot. pub fn mesh(&self) -> Option<&crate::mesh_boot::MeshHandle> { diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 8a4b0c6d665..b022dd8a27f 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -115,6 +115,27 @@ 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. +### Readiness telemetry contract + +Only requests served by the private health listener (`BUZZ_HEALTH_PORT`) emit +rollout readiness telemetry. The compatibility `/_readiness` route on the public +app listener returns health but does not change these metrics. + +| Metric | Type | Labels | +|--------|------|--------| +| `buzz_readiness_checks_total` | counter | `reason` from the closed readiness-reason set | +| `buzz_readiness_dependency_checks_total` | counter | `dependency`, typed bounded `outcome` | +| `buzz_readiness_check_duration_seconds` | histogram | `check` only | +| `buzz_readiness_state` | gauge | `check` only; latest publishable generation | + +The schema has a ceiling of 99 raw Prometheus series per pod: 12 overall +reasons, 11 valid dependency/outcome pairs, 72 histogram series, and 4 gauges. +Do not add pod, ReplicaSet, version, rollout, error text, SQL, URL, tenant, +user, community, pubkey, header, query, or other request-controlled labels. +Shutdown without dependency evaluation increments only +`buzz_readiness_checks_total{reason="shutting_down"}` and sets the overall +state to zero; it does not fabricate dependency failures or latency samples. + ## Relay Pod extensions The chart exposes narrow extension points for init containers, volumes, relay