feat(relay): add detailed readiness metrics

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,

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)

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());

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.

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;

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: