diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 625eb50737f..5c972fb551e 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -31,8 +31,8 @@ pub mod error; mod test_support; pub use runtime::{ - insert_mentions, migration, replica_fence, Db, DbConfig, DbPoolRole, DbPoolStats, - DbReadinessOutcome, ReadSession, + insert_mentions, migration, replica_fence, Db, DbConfig, DbConnectionOutcome, DbConnectionStep, + DbPoolRole, DbPoolStats, DbReadinessOutcome, ReadSession, }; /// Valid low-cardinality `(pool_role, operation)` pairs for pool-acquisition telemetry. @@ -42,6 +42,21 @@ pub const DB_POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 11] = /// Raw Prometheus series ceiling per relay pod for the operation-aware contract. pub const DB_POOL_ACQUIRE_RAW_SERIES_PER_POD: usize = runtime::observability::POOL_ACQUIRE_RAW_SERIES_PER_POD; + +/// Valid database connection role/step pairs with a start counter. +pub const DB_CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 4] = + runtime::CONNECTION_STARTED_STEPS; + +/// Valid database connection role/step pairs with a duration histogram. +pub const DB_CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 4] = + runtime::CONNECTION_DURATION_STEPS; + +/// Valid database connection role/step/outcome terminal combinations. +pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 15] = + runtime::CONNECTION_TERMINALS; + +/// Raw Prometheus series ceiling per pod for connection-step telemetry. +pub const DB_CONNECTION_RAW_SERIES_PER_POD: usize = runtime::CONNECTION_RAW_SERIES_PER_POD; pub(crate) use runtime::{ insert_mentions_in_transaction, observability, route_proof, ReadSessionInner, RouteDecision, RoutePredicate, diff --git a/crates/buzz-db/src/runtime/connection_observability.rs b/crates/buzz-db/src/runtime/connection_observability.rs new file mode 100644 index 00000000000..a1ab04022fd --- /dev/null +++ b/crates/buzz-db/src/runtime/connection_observability.rs @@ -0,0 +1,344 @@ +//! Fixed-vocabulary metrics for writer-pool connection setup. +//! +//! SQLx exposes the point immediately after a physical connection succeeds, +//! but it does not expose a callback immediately before each physical dial. +//! Consequently, `physical_connect` is a success milestone rather than a +//! duration phase. The aggregate `writer_pool` phase owns failures that occur +//! before `after_connect`, while the session phases own their exact failures. + +use std::time::{Duration, Instant}; + +use super::DbPoolRole; + +/// Fixed writer connection-setup steps. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DbConnectionStep { + /// Construct the writer pool and satisfy its initial minimum size. + WriterPool, + /// A physical connection has completed DNS/network/TLS/authentication. + PhysicalConnect, + /// Install the created-at replica-fence floor. + CreatedAtFloor, + /// Install lock, idle-transaction, and statement timeouts. + SessionTimeouts, + /// Verify READ COMMITTED transaction isolation. + Isolation, + /// The physical connection passed every required session premise. + Ready, +} + +impl DbConnectionStep { + /// Complete metric-label vocabulary. + pub const ALL: [Self; 6] = [ + Self::WriterPool, + Self::PhysicalConnect, + Self::CreatedAtFloor, + Self::SessionTimeouts, + Self::Isolation, + Self::Ready, + ]; + + /// Stable metric label. + pub const fn as_str(self) -> &'static str { + match self { + Self::WriterPool => "writer_pool", + Self::PhysicalConnect => "physical_connect", + Self::CreatedAtFloor => "created_at_floor", + Self::SessionTimeouts => "session_timeouts", + Self::Isolation => "isolation", + Self::Ready => "ready", + } + } +} + +/// Bounded terminal outcome for a connection-setup step. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DbConnectionOutcome { + /// The step completed successfully. + Succeeded, + /// The step failed. + Failed, + /// The aggregate pool deadline expired. + TimedOut, + /// The owning future was dropped before a terminal. + Cancelled, +} + +impl DbConnectionOutcome { + /// Stable metric label. + pub const fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "succeeded", + Self::Failed => "failed", + Self::TimedOut => "timed_out", + Self::Cancelled => "cancelled", + } + } +} + +/// Valid role/step pairs with an explicit start counter. +pub const CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 4] = [ + (DbPoolRole::Writer, DbConnectionStep::WriterPool), + (DbPoolRole::Writer, DbConnectionStep::CreatedAtFloor), + (DbPoolRole::Writer, DbConnectionStep::SessionTimeouts), + (DbPoolRole::Writer, DbConnectionStep::Isolation), +]; + +/// Valid role/step pairs with a duration histogram. +pub const CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 4] = CONNECTION_STARTED_STEPS; + +/// Valid role/step/outcome terminal combinations. +pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 15] = [ + ( + DbPoolRole::Writer, + DbConnectionStep::WriterPool, + DbConnectionOutcome::Succeeded, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::WriterPool, + DbConnectionOutcome::Failed, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::WriterPool, + DbConnectionOutcome::TimedOut, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::WriterPool, + DbConnectionOutcome::Cancelled, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::PhysicalConnect, + DbConnectionOutcome::Succeeded, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::CreatedAtFloor, + DbConnectionOutcome::Succeeded, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::CreatedAtFloor, + DbConnectionOutcome::Failed, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::CreatedAtFloor, + DbConnectionOutcome::Cancelled, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::SessionTimeouts, + DbConnectionOutcome::Succeeded, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::SessionTimeouts, + DbConnectionOutcome::Failed, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::SessionTimeouts, + DbConnectionOutcome::Cancelled, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::Isolation, + DbConnectionOutcome::Succeeded, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::Isolation, + DbConnectionOutcome::Failed, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::Isolation, + DbConnectionOutcome::Cancelled, + ), + ( + DbPoolRole::Writer, + DbConnectionStep::Ready, + DbConnectionOutcome::Succeeded, + ), +]; + +/// Four start counters + four 13-series histograms + fifteen terminal counters. +pub const CONNECTION_RAW_SERIES_PER_POD: usize = 4 + (4 * 13) + 15; + +pub(crate) struct DbConnectionStepAttempt { + pool_role: DbPoolRole, + step: DbConnectionStep, + started: Instant, + finished: bool, +} + +impl DbConnectionStepAttempt { + pub(crate) fn start(pool_role: DbPoolRole, step: DbConnectionStep) -> Self { + metrics::counter!( + "buzz_db_connection_step_started_total", + "pool_role" => pool_role.as_str(), + "step" => step.as_str(), + ) + .increment(1); + Self { + pool_role, + step, + started: Instant::now(), + finished: false, + } + } + + pub(crate) fn succeed(self) { + self.finish(DbConnectionOutcome::Succeeded); + } + + pub(crate) fn fail(self) { + self.finish(DbConnectionOutcome::Failed); + } + + pub(crate) fn time_out(self) { + self.finish(DbConnectionOutcome::TimedOut); + } + + fn finish(mut self, outcome: DbConnectionOutcome) { + record_terminal( + self.pool_role, + self.step, + outcome, + Some(self.started.elapsed()), + ); + self.finished = true; + } +} + +impl Drop for DbConnectionStepAttempt { + fn drop(&mut self) { + if self.finished { + return; + } + let outcome = if std::thread::panicking() { + DbConnectionOutcome::Failed + } else { + DbConnectionOutcome::Cancelled + }; + record_terminal( + self.pool_role, + self.step, + outcome, + Some(self.started.elapsed()), + ); + self.finished = true; + } +} + +pub(crate) fn record_milestone(pool_role: DbPoolRole, step: DbConnectionStep) { + record_terminal(pool_role, step, DbConnectionOutcome::Succeeded, None); +} + +fn record_terminal( + pool_role: DbPoolRole, + step: DbConnectionStep, + outcome: DbConnectionOutcome, + elapsed: Option, +) { + metrics::counter!( + "buzz_db_connection_step_attempts_total", + "pool_role" => pool_role.as_str(), + "step" => step.as_str(), + "outcome" => outcome.as_str(), + ) + .increment(1); + if let Some(elapsed) = elapsed { + metrics::histogram!( + "buzz_db_connection_step_duration_seconds", + "pool_role" => pool_role.as_str(), + "step" => step.as_str(), + ) + .record(elapsed.as_secs_f64()); + } +} + +pub(crate) fn classify_pool_outcome(error: &sqlx::Error) -> DbConnectionOutcome { + if matches!(error, sqlx::Error::PoolTimedOut) { + DbConnectionOutcome::TimedOut + } else { + DbConnectionOutcome::Failed + } +} + +#[cfg(test)] +mod tests { + use super::*; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + #[test] + fn vocabulary_and_series_budget_are_frozen() { + assert_eq!( + DbConnectionStep::ALL.map(DbConnectionStep::as_str), + [ + "writer_pool", + "physical_connect", + "created_at_floor", + "session_timeouts", + "isolation", + "ready", + ] + ); + assert_eq!(CONNECTION_RAW_SERIES_PER_POD, 71); + } + + #[test] + fn dropped_step_is_cancelled_exactly_once_without_sensitive_labels() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + let attempt = + DbConnectionStepAttempt::start(DbPoolRole::Writer, DbConnectionStep::CreatedAtFloor); + drop(attempt); + + let metrics = snapshotter.snapshot().into_vec(); + assert_eq!(metrics.len(), 3); + for (key, _, _, value) in metrics { + let labels = key + .key() + .labels() + .map(|label| (label.key(), label.value())) + .collect::>(); + assert_eq!(labels.get("pool_role"), Some(&"writer")); + assert_eq!(labels.get("step"), Some(&"created_at_floor")); + assert!(!labels.contains_key("reason")); + assert!(!labels.contains_key("connection_ordinal")); + match value { + DebugValue::Counter(value) => { + assert_eq!(value, 1); + if key.key().name() == "buzz_db_connection_step_attempts_total" { + assert_eq!(labels.get("outcome"), Some(&"cancelled")); + } + } + DebugValue::Histogram(values) => assert_eq!(values.len(), 1), + DebugValue::Gauge(_) => panic!("connection setup has no gauges"), + } + } + } + + #[test] + fn pool_timeout_is_distinguished_from_other_failures() { + assert_eq!( + classify_pool_outcome(&sqlx::Error::PoolTimedOut), + DbConnectionOutcome::TimedOut + ); + assert_eq!( + classify_pool_outcome(&sqlx::Error::PoolClosed), + DbConnectionOutcome::Failed + ); + let io = sqlx::Error::Io(std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "postgres://secret-user:secret-password@example.invalid/private", + )); + assert_eq!(classify_pool_outcome(&io), DbConnectionOutcome::Failed); + } +} diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 00bb0dc3259..150c8f47682 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -1,7 +1,14 @@ +mod connection_observability; pub mod migration; pub(crate) mod observability; pub mod replica_fence; +pub use connection_observability::{DbConnectionOutcome, DbConnectionStep}; +pub(crate) use connection_observability::{ + CONNECTION_DURATION_STEPS, CONNECTION_RAW_SERIES_PER_POD, CONNECTION_STARTED_STEPS, + CONNECTION_TERMINALS, +}; + use crate::{deletion, event, DbError, EventQuery, Result}; use buzz_datastore_tracing::datastore_span; use chrono::{DateTime, Utc}; @@ -630,6 +637,10 @@ impl Db { /// constructor so they inherit the timeout, floor-guard, and isolation /// policy installed by [`Db::new`]. pub async fn connect_writer_pool(config: &DbConfig) -> Result { + use connection_observability::{ + classify_pool_outcome, record_milestone, DbConnectionStep, DbConnectionStepAttempt, + }; + let lock_timeout_ms = config.lock_timeout_ms; let idle_txn_timeout_ms = config.idle_txn_timeout_ms; let statement_timeout_ms = config.statement_timeout_ms; @@ -641,11 +652,27 @@ impl Db { .idle_timeout(Duration::from_secs(config.idle_timeout_secs)) .after_connect(move |conn, _meta| { Box::pin(async move { + // SQLx 0.9 exposes no callback immediately before each raw + // physical dial. Entering `after_connect` is the truthful + // point at which DNS/network/TLS/authentication succeeded. + record_milestone(DbPoolRole::Writer, DbConnectionStep::PhysicalConnect); + // `SET` cannot take bind parameters; `set_config` can. - sqlx::query("SELECT set_config('buzz.created_at_floor', $1, false)") + let floor = DbConnectionStepAttempt::start( + DbPoolRole::Writer, + DbConnectionStep::CreatedAtFloor, + ); + if let Err(error) = + sqlx::query("SELECT set_config('buzz.created_at_floor', $1, false)") .bind(replica_fence::CREATED_AT_FLOOR_SECS.to_string()) .execute(&mut *conn) - .await?; + .await + { + floor.fail(); + return Err(error); + } + floor.succeed(); + // `lock_timeout` fails the waiting statement; it does not // cancel the holder. `idle_in_transaction_session_timeout` // reaps only holders idling inside an open transaction, @@ -654,7 +681,11 @@ impl Db { // milliseconds. Migration/schema-destruction connections // reset lock and statement timeouts before their intentional // long wait (see `with_exclusive_schema_destruction_lock`). - sqlx::query( + let timeouts = DbConnectionStepAttempt::start( + DbPoolRole::Writer, + DbConnectionStep::SessionTimeouts, + ); + if let Err(error) = sqlx::query( "SELECT set_config('lock_timeout', $1, false), \ set_config('idle_in_transaction_session_timeout', $2, false), \ set_config('statement_timeout', $3, false)", @@ -663,11 +694,29 @@ impl Db { .bind(idle_txn_timeout_ms.to_string()) .bind(statement_timeout_ms.to_string()) .execute(&mut *conn) - .await?; - let isolation: String = sqlx::query_scalar("SHOW transaction_isolation") + .await + { + timeouts.fail(); + return Err(error); + } + timeouts.succeed(); + + let isolation_step = DbConnectionStepAttempt::start( + DbPoolRole::Writer, + DbConnectionStep::Isolation, + ); + let isolation: String = match sqlx::query_scalar("SHOW transaction_isolation") .fetch_one(&mut *conn) - .await?; + .await + { + Ok(isolation) => isolation, + Err(error) => { + isolation_step.fail(); + return Err(error); + } + }; if isolation != "read committed" { + isolation_step.fail(); return Err(sqlx::Error::Configuration( format!( "writer pool requires READ COMMITTED transaction isolation, got {isolation}" @@ -675,10 +724,29 @@ impl Db { .into(), )); } + isolation_step.succeed(); + record_milestone(DbPoolRole::Writer, DbConnectionStep::Ready); Ok(()) }) }); - Ok(options.connect(&config.database_url).await?) + + let pool_attempt = + DbConnectionStepAttempt::start(DbPoolRole::Writer, DbConnectionStep::WriterPool); + match options.connect(&config.database_url).await { + Ok(pool) => { + pool_attempt.succeed(); + Ok(pool) + } + Err(error) => { + let outcome = classify_pool_outcome(&error); + if outcome == connection_observability::DbConnectionOutcome::TimedOut { + pool_attempt.time_out(); + } else { + pool_attempt.fail(); + } + Err(error.into()) + } + } } /// Reader acquire timeout — deliberately far below the writer's diff --git a/crates/buzz-db/src/runtime/observability.rs b/crates/buzz-db/src/runtime/observability.rs index 334ed60255d..079c43f98a7 100644 --- a/crates/buzz-db/src/runtime/observability.rs +++ b/crates/buzz-db/src/runtime/observability.rs @@ -153,8 +153,8 @@ pub(crate) const POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 11] = [ (DbPoolRole::Writer.as_str(), "maintenance"), ]; -/// Eleven valid pairs × (12 histogram series + 4 outcome counters + 1 gauge). -pub(crate) const POOL_ACQUIRE_RAW_SERIES_PER_POD: usize = POOL_ACQUIRE_VALID_PAIRS.len() * 17; +/// Eleven valid pairs × (12 histogram series + 1 start counter + 4 outcome counters + 1 gauge). +pub(crate) const POOL_ACQUIRE_RAW_SERIES_PER_POD: usize = POOL_ACQUIRE_VALID_PAIRS.len() * 18; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum LockType { @@ -382,6 +382,12 @@ struct PoolAcquireAttempt { impl PoolAcquireAttempt { fn start(pair: PoolOperation, emit_legacy: bool) -> Self { + metrics::counter!( + "buzz_db_pool_acquire_started_total", + "pool_role" => pair.pool_role(), + "operation" => pair.operation(), + ) + .increment(1); { let mut waiters = POOL_WAITERS[pair.index()] .lock() @@ -606,7 +612,7 @@ mod tests { PoolOperation::ReaderSubscriptionHistory, ] ); - assert_eq!(super::POOL_ACQUIRE_RAW_SERIES_PER_POD, 187); + assert_eq!(super::POOL_ACQUIRE_RAW_SERIES_PER_POD, 198); assert_eq!( LockType::ALL.map(LockType::as_str), [ @@ -863,6 +869,24 @@ mod tests { let _guard = metrics::set_default_local_recorder(&recorder); let attempt = PoolAcquireAttempt::start(PoolOperation::WriterTenantResolution, false); + let in_flight = snapshotter.snapshot().into_vec(); + assert!(in_flight.iter().any(|(key, _, _, value)| { + let labels = key + .key() + .labels() + .map(|label| (label.key(), label.value())) + .collect::>(); + matches!(value, DebugValue::Counter(1)) + && key.key().name() == "buzz_db_pool_acquire_started_total" + && labels.get("pool_role") == Some(&"writer") + && labels.get("operation") == Some(&"tenant_resolution") + })); + assert!( + in_flight.iter().all(|(key, _, _, _)| { + key.key().name() != "buzz_db_pool_acquire_attempts_total" + }), + "the request-start signal must be observable before its terminal" + ); drop(attempt); refresh_pool_waiters(true); diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index d97ebf0ac54..cce1927be69 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -1,11 +1,69 @@ use super::*; use crate::{relay_members, thread}; use buzz_core::CommunityId; +use metrics_util::debugging::{DebugValue, DebuggingRecorder, Snapshotter}; use sqlx::{Connection, PgPool}; use uuid::Uuid; const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials +type ConnectionCounters = std::collections::BTreeMap<(String, String, Option), u64>; + +fn connection_counters(snapshotter: &Snapshotter) -> ConnectionCounters { + snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(key, _, _, value)| { + let metric_name = key.key().name(); + if ![ + "buzz_db_connection_step_started_total", + "buzz_db_connection_step_attempts_total", + ] + .contains(&metric_name) + { + return None; + } + let DebugValue::Counter(value) = value else { + panic!("{metric_name} must be a counter"); + }; + let labels = key + .key() + .labels() + .map(|label| (label.key().to_owned(), label.value().to_owned())) + .collect::>(); + assert_eq!(labels.get("pool_role").map(String::as_str), Some("writer")); + Some(( + ( + metric_name.to_owned(), + labels + .get("step") + .expect("connection step label") + .to_owned(), + labels.get("outcome").cloned(), + ), + value, + )) + }) + .collect() +} + +fn connection_counter( + counters: &ConnectionCounters, + metric_name: &str, + step: DbConnectionStep, + outcome: Option, +) -> u64 { + counters + .get(&( + metric_name.to_owned(), + step.as_str().to_owned(), + outcome.map(|outcome| outcome.as_str().to_owned()), + )) + .copied() + .unwrap_or_default() +} + async fn setup_db() -> Db { let database_url = std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into()); let pool = PgPool::connect(&database_url) @@ -2544,9 +2602,151 @@ fn writer_pool_safety_hook_is_single_and_composed() { assert!(!reader_doc.contains("Db::connect_writer_pool")); } -#[tokio::test] +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn writer_pool_metrics_record_every_initial_connection_ready() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + let db = Db::new(&DbConfig { + database_url: crate::test_support::database_url(), + max_connections: 2, + min_connections: 2, + ..DbConfig::default() + }) + .await + .expect("connect instrumented writer pool"); + let counters = connection_counters(&snapshotter); + + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_started_total", + DbConnectionStep::WriterPool, + None, + ), + 1 + ); + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::WriterPool, + Some(DbConnectionOutcome::Succeeded), + ), + 1 + ); + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::PhysicalConnect, + Some(DbConnectionOutcome::Succeeded), + ), + 2 + ); + for step in [ + DbConnectionStep::CreatedAtFloor, + DbConnectionStep::SessionTimeouts, + DbConnectionStep::Isolation, + ] { + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_started_total", + step, + None, + ), + 2, + "every initial connection must start {}", + step.as_str(), + ); + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_attempts_total", + step, + Some(DbConnectionOutcome::Succeeded), + ), + 2, + "every initial connection must complete {}", + step.as_str(), + ); + } + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::Ready, + Some(DbConnectionOutcome::Succeeded), + ), + 2 + ); + db.pool.close().await; +} + +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn writer_pool_metrics_record_connection_created_after_startup() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + let pool = Db::connect_writer_pool(&DbConfig { + database_url: crate::test_support::database_url(), + max_connections: 2, + min_connections: 1, + ..DbConfig::default() + }) + .await + .expect("connect instrumented size-one writer pool"); + let startup_counters = connection_counters(&snapshotter); + assert_eq!( + connection_counter( + &startup_counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::Ready, + Some(DbConnectionOutcome::Succeeded), + ), + 1 + ); + + let first = pool.acquire().await.expect("hold initial connection"); + let second = pool + .acquire() + .await + .expect("grow pool with a second connection"); + let growth_counters = connection_counters(&snapshotter); + assert_eq!( + connection_counter( + &growth_counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::Ready, + Some(DbConnectionOutcome::Succeeded), + ), + 1, + "a post-startup pool growth connection must traverse the same instrumented safety hook" + ); + assert_eq!( + connection_counter( + &growth_counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::PhysicalConnect, + Some(DbConnectionOutcome::Succeeded), + ), + 1, + ); + + drop(second); + drop(first); + pool.close().await; +} + +#[tokio::test(flavor = "current_thread")] #[ignore = "requires Postgres"] async fn writer_pool_rejects_non_read_committed_database_default() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); let admin = PgPool::connect(&admin_url().await) .await .expect("connect admin"); @@ -2576,6 +2776,26 @@ async fn writer_pool_rejects_non_read_committed_database_default() { || error.to_string().contains("pool timed out"), "unexpected isolation rejection: {error}" ); + let counters = connection_counters(&snapshotter); + assert!( + connection_counter( + &counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::Isolation, + Some(DbConnectionOutcome::Failed), + ) > 0, + "the failing production hook must record the isolation failure" + ); + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::Ready, + Some(DbConnectionOutcome::Succeeded), + ), + 0, + "an isolation-rejected connection must never reach ready" + ); sqlx::query(sqlx::AssertSqlSafe(format!( "DROP DATABASE {name} WITH (FORCE)" @@ -2585,6 +2805,59 @@ async fn writer_pool_rejects_non_read_committed_database_default() { .expect("drop isolation test database"); } +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn writer_pool_metrics_stop_after_session_timeout_setup_failure() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + let error = Db::new(&DbConfig { + database_url: crate::test_support::database_url(), + max_connections: 1, + min_connections: 1, + acquire_timeout_secs: 1, + statement_timeout_ms: u64::MAX, + ..DbConfig::default() + }) + .await + .expect_err("Postgres must reject an out-of-range statement timeout"); + assert!( + error.to_string().contains("pool timed out") + || error.to_string().contains("invalid value for parameter") + ); + let counters = connection_counters(&snapshotter); + + assert!( + connection_counter( + &counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::SessionTimeouts, + Some(DbConnectionOutcome::Failed), + ) > 0, + "the failing production hook must record the timeout-setup failure" + ); + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_started_total", + DbConnectionStep::Isolation, + None, + ), + 0, + "a timeout-setup failure must stop before isolation" + ); + assert_eq!( + connection_counter( + &counters, + "buzz_db_connection_step_attempts_total", + DbConnectionStep::Ready, + Some(DbConnectionOutcome::Succeeded), + ), + 0, + "a timeout-setup failure must never reach ready" + ); +} + /// Session-timeout environment overrides retain PostgreSQL's `0 = disabled` /// semantics and ignore invalid values. #[test] diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index cb45809eadb..a6c3eae8f0a 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -237,6 +237,11 @@ pub async fn huddle_started_links( if parent_channel_ids.is_empty() || ephemeral_channel_ids.is_empty() { return Ok(Vec::new()); } + let mut connection = crate::observability::acquire_writer( + pool, + crate::observability::WriterOperation::SubscriptionHistory, + ) + .await?; let rows = sqlx::query( r#" SELECT DISTINCT ON (backing.id) @@ -267,7 +272,7 @@ pub async fn huddle_started_links( .bind(KIND_HUDDLE_STARTED as i32) .bind(ephemeral_channel_ids) .bind(HUDDLE_LINK_CONTENT_MAX_BYTES) - .fetch_all(pool) + .fetch_all(&mut *connection) .await?; rows.into_iter() @@ -2796,7 +2801,7 @@ mod postgres_tests { ); } - #[tokio::test] + #[tokio::test(flavor = "current_thread")] #[ignore = "requires Postgres"] async fn huddle_started_links_batches_valid_creator_links_and_ignores_malformed_content() { let pool = setup_pool().await; @@ -2831,10 +2836,60 @@ mod postgres_tests { .expect("insert huddle-start candidate"); } - let links = huddle_started_links(&pool, community, &[parent], &[session]) - .await - .expect("batch huddle links"); + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let db = Db::from_pool(pool); + let links = { + let _guard = metrics::set_default_local_recorder(&recorder); + db.huddle_started_links(community, &[parent], &[session]) + .await + } + .expect("batch huddle links"); assert_eq!(links, vec![(session, parent, creator)]); + + let counters = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(key, _, _, value)| { + let labels = key + .key() + .labels() + .map(|label| (label.key().to_owned(), label.value().to_owned())) + .collect::>(); + if labels.get("pool_role").map(String::as_str) != Some("writer") + || labels.get("operation").map(String::as_str) != Some("subscription_history") + { + return None; + } + let metrics_util::debugging::DebugValue::Counter(value) = value else { + return None; + }; + Some(( + (key.key().name().to_owned(), labels.get("outcome").cloned()), + value, + )) + }) + .collect::>(); + assert_eq!( + counters, + [ + ( + ("buzz_db_pool_acquire_started_total".to_owned(), None), + 1, + ), + ( + ( + "buzz_db_pool_acquire_attempts_total".to_owned(), + Some("success".to_owned()), + ), + 1, + ), + ] + .into_iter() + .collect(), + "the production huddle lookup must emit one writer/subscription_history start and success terminal" + ); } #[test] diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 761b821d5ef..ef28ee0f3c3 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -43,6 +43,12 @@ const DB_POOL_ACQUIRE_DURATION_BUCKETS_S: [f64; 9] = [0.001, 0.005, 0.01, 0.025, 0.05, 0.15, 0.5, 1.0, 3.0]; const DB_POOL_ACQUIRE_DURATION_UNIT: metrics::Unit = metrics::Unit::Seconds; +/// Writer pool/session buckets preserve sub-millisecond setup while retaining +/// seconds-scale failures in the final bucket. +const DB_CONNECTION_STEP_DURATION_BUCKETS_S: [f64; 10] = [ + 0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.15, 0.5, 1.0, 3.0, +]; + /// 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, @@ -115,6 +121,11 @@ fn configured_prometheus_builder(gauge_idle_timeout_secs: u64) -> PrometheusBuil &DB_POOL_ACQUIRE_DURATION_BUCKETS_S, ) .expect("valid DB pool acquisition duration bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_db_connection_step_duration_seconds".to_owned()), + &DB_CONNECTION_STEP_DURATION_BUCKETS_S, + ) + .expect("valid DB connection-step duration bucket boundaries") .set_buckets_for_metric( Matcher::Full("buzz_git_hydrate_bytes".to_owned()), &GIT_BYTES_BUCKETS, @@ -242,6 +253,10 @@ pub(crate) fn describe_readiness_metrics() { /// Register the frozen operation-aware pool-acquisition contract. pub(crate) fn describe_db_pool_metrics() { + metrics::describe_counter!( + "buzz_db_pool_acquire_started_total", + "Database pool checkout starts by valid pool role and operation" + ); metrics::describe_histogram!( "buzz_db_pool_acquire_duration_seconds", DB_POOL_ACQUIRE_DURATION_UNIT, @@ -255,6 +270,19 @@ pub(crate) fn describe_db_pool_metrics() { "buzz_db_pool_waiters", "Current tracked-operation database pool checkout attempts in progress by valid pool role and operation" ); + metrics::describe_counter!( + "buzz_db_connection_step_started_total", + "Writer connection setup phase starts by fixed pool role and step" + ); + metrics::describe_counter!( + "buzz_db_connection_step_attempts_total", + "Writer connection setup terminals by fixed pool role, step, and outcome" + ); + metrics::describe_histogram!( + "buzz_db_connection_step_duration_seconds", + metrics::Unit::Seconds, + "Writer connection setup phase duration by fixed pool role and step" + ); metrics::describe_gauge!( "buzz_db_pool_connections", "Current Postgres pool connections by physical pool role and bounded utilization state" @@ -633,11 +661,17 @@ mod contract_tests { } #[test] - fn production_builder_exports_frozen_db_pool_contract_and_187_series_budget() { + fn production_builder_exports_frozen_db_pool_and_connection_contracts() { let (recorder, handle) = super::readiness_test_recorder(); metrics::with_local_recorder(&recorder, || { super::describe_db_pool_metrics(); for (pool_role, operation) in buzz_db::DB_POOL_ACQUIRE_VALID_PAIRS { + metrics::counter!( + "buzz_db_pool_acquire_started_total", + "pool_role" => pool_role, + "operation" => operation, + ) + .increment(1); metrics::histogram!( "buzz_db_pool_acquire_duration_seconds", "pool_role" => pool_role, @@ -660,15 +694,44 @@ mod contract_tests { .increment(1); } } + for (pool_role, step) in buzz_db::DB_CONNECTION_STARTED_STEPS { + metrics::counter!( + "buzz_db_connection_step_started_total", + "pool_role" => pool_role.as_str(), + "step" => step.as_str(), + ) + .increment(1); + } + for (pool_role, step) in buzz_db::DB_CONNECTION_DURATION_STEPS { + metrics::histogram!( + "buzz_db_connection_step_duration_seconds", + "pool_role" => pool_role.as_str(), + "step" => step.as_str(), + ) + .record(0.02); + } + for (pool_role, step, outcome) in buzz_db::DB_CONNECTION_TERMINALS { + metrics::counter!( + "buzz_db_connection_step_attempts_total", + "pool_role" => pool_role.as_str(), + "step" => step.as_str(), + "outcome" => outcome.as_str(), + ) + .increment(1); + } }); let scrape = handle.render(); + assert!(scrape.contains("# TYPE buzz_db_pool_acquire_started_total counter")); assert!(scrape.contains("# TYPE buzz_db_pool_acquire_duration_seconds histogram")); assert!(scrape.contains("# TYPE buzz_db_pool_acquire_attempts_total counter")); assert!(scrape.contains("# TYPE buzz_db_pool_waiters gauge")); assert!(scrape.contains("# HELP buzz_db_pool_acquire_duration_seconds Database pool checkout duration by valid pool role and operation")); assert!(scrape.contains("# HELP buzz_db_pool_acquire_attempts_total Database pool checkout terminals by valid pool role, operation, and outcome")); assert!(scrape.contains("# HELP buzz_db_pool_waiters Current tracked-operation database pool checkout attempts in progress by valid pool role and operation")); + assert!(scrape.contains("# TYPE buzz_db_connection_step_started_total counter")); + assert!(scrape.contains("# TYPE buzz_db_connection_step_attempts_total counter")); + assert!(scrape.contains("# TYPE buzz_db_connection_step_duration_seconds histogram")); assert_eq!(super::DB_POOL_ACQUIRE_DURATION_UNIT, metrics::Unit::Seconds); let readiness_buckets = scrape .lines() @@ -693,7 +756,8 @@ mod contract_tests { let raw_series = scrape .lines() .filter(|line| { - line.starts_with("buzz_db_pool_acquire_duration_seconds") + line.starts_with("buzz_db_pool_acquire_started_total") + || line.starts_with("buzz_db_pool_acquire_duration_seconds") || line.starts_with("buzz_db_pool_acquire_attempts_total") || line.starts_with("buzz_db_pool_waiters{") }) @@ -708,7 +772,9 @@ mod contract_tests { let keys = label_keys(line); if line.starts_with("buzz_db_pool_acquire_duration_seconds_bucket") { assert_eq!(keys, BTreeSet::from(["le", "operation", "pool_role"])); - } else if line.starts_with("buzz_db_pool_acquire_duration_seconds") { + } else if line.starts_with("buzz_db_pool_acquire_duration_seconds") + || line.starts_with("buzz_db_pool_acquire_started_total") + { assert_eq!(keys, BTreeSet::from(["operation", "pool_role"])); } else if line.starts_with("buzz_db_pool_acquire_attempts_total") { assert_eq!(keys, BTreeSet::from(["operation", "outcome", "pool_role"])); @@ -718,6 +784,32 @@ mod contract_tests { assert!(!line.contains("operation=\"other\"")); assert!(!line.contains("result=")); } + + let connection_series = scrape + .lines() + .filter(|line| { + line.starts_with("buzz_db_connection_step_started_total") + || line.starts_with("buzz_db_connection_step_attempts_total") + || line.starts_with("buzz_db_connection_step_duration_seconds") + }) + .collect::>(); + assert_eq!( + connection_series.len(), + buzz_db::DB_CONNECTION_RAW_SERIES_PER_POD, + "unexpected DB connection scrape:\n{scrape}" + ); + for line in connection_series { + let keys = label_keys(line); + if line.starts_with("buzz_db_connection_step_duration_seconds_bucket") { + assert_eq!(keys, BTreeSet::from(["le", "pool_role", "step"])); + } else if line.starts_with("buzz_db_connection_step_attempts_total") { + assert_eq!(keys, BTreeSet::from(["outcome", "pool_role", "step"])); + } else { + assert_eq!(keys, BTreeSet::from(["pool_role", "step"])); + } + assert!(!line.contains("reason=")); + assert!(!line.contains("connection_ordinal=")); + } } } diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 91a2b59ea3d..a6b74e87572 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -148,8 +148,9 @@ state to zero; it does not fabricate dependency failures or latency samples. ### Operation-aware database pool acquisition contract -The operation-aware families separate three questions: who is waiting now, -how completed/abandoned attempts ended, and how long checkout waits took. +The operation-aware families separate four questions: when an operation asked +for a connection, who is waiting now, how completed/abandoned attempts ended, +and how long checkout waits took. Outcome remains on the terminal counter for historical deployment comparison; it is intentionally absent from the expensive duration histogram. @@ -161,6 +162,7 @@ maximum gauges when diagnosing total capacity pressure. | Metric | Type | Labels | |--------|------|--------| +| `buzz_db_pool_acquire_started_total` | counter | `pool_role`, `operation` | | `buzz_db_pool_acquire_duration_seconds` | histogram | `pool_role`, `operation` | | `buzz_db_pool_acquire_attempts_total` | counter | `pool_role`, `operation`, `outcome` | | `buzz_db_pool_waiters` | gauge | `pool_role`, `operation`; tracked operations only, periodically refreshed including zero | @@ -182,12 +184,49 @@ writer/maintenance ``` Nine finite checkout buckets plus `+Inf`, sum, and count yield 12 histogram -series per valid pair. The new contract therefore has a hard ceiling of 187 -raw Prometheus series per pod: `11 × (12 + 4 + 1)`. The two legacy acquisition +series per valid pair. The new contract therefore has a hard ceiling of 198 +raw Prometheus series per pod: `11 × (1 + 12 + 4 + 1)`. The two legacy acquisition families remain temporarily for dashboard compatibility and are not part of that new-family budget. No `other` operation or request-controlled/sensitive label is valid. +### Writer connection setup contract + +Writer connection setup is separate from checkout. At boot, SQLx constructs +the writer pool and creates its minimum physical connections. Later it may +create more when the pool grows or replaces a broken or expired connection. +Every connected writer session must install the created-at floor, install the +session timeouts, verify READ COMMITTED isolation, and reach `ready` before SQLx +can give it to a caller. + +| Metric | Type | Labels | +|--------|------|--------| +| `buzz_db_connection_step_started_total` | counter | `pool_role`, `step` | +| `buzz_db_connection_step_duration_seconds` | histogram | `pool_role`, `step` | +| `buzz_db_connection_step_attempts_total` | counter | `pool_role`, `step`, `outcome` | + +The fixed steps are `writer_pool`, `physical_connect`, `created_at_floor`, +`session_timeouts`, `isolation`, and `ready`. Measurable phases emit start, +duration, and terminal evidence. `physical_connect` and `ready` are success +milestones: SQLx 0.9 exposes `after_connect` only after DNS, network, TLS, and +authentication finish, so Buzz does not invent separate timings for those +internal phases. Raw connect failures before `after_connect` are classified on +the aggregate `writer_pool` phase during initial construction. Outcomes are +`succeeded`, `failed`, `timed_out`, and `cancelled` where valid. + +For a measurable phase on one pod, subtract all terminal outcomes from its +start counter to derive the number of attempts currently in progress. The +session phases run in order, so a later phase start also proves the earlier +phases succeeded for that attempt. + +Four start counters, four 13-series histograms, and fifteen terminal counters +create a hard ceiling of 71 raw Prometheus series per pod. Connection ordinals, +database URLs, hosts, usernames, SQL, and raw errors are forbidden as metric +labels. + +When audit logging is enabled, the `buzz_db_connection_*` metric totals combine +the main writer pool and the separate audit writer pool. + ### Physical database pool utilization and configuration contract Every physical Postgres pool the relay owns reports current utilization,