diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 6f8d0ffb3d4..6fe3c245d52 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -31,8 +31,9 @@ pub mod error; mod test_support; pub use runtime::{ - insert_mentions, migration, replica_fence, Db, DbConfig, DbPoolStats, DbReadinessOutcome, - ReadSession, + insert_mentions, migration, replica_fence, Db, DbConfig, DbConnectionEdge, + DbConnectionLifecycleEvent, DbConnectionObserver, DbConnectionOutcome, DbConnectionReason, + DbConnectionStep, DbPoolRole, DbPoolStats, DbReadinessOutcome, ReadSession, }; /// Valid low-cardinality `(pool_role, operation)` pairs for pool-acquisition telemetry. @@ -42,6 +43,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..8a90a498b40 --- /dev/null +++ b/crates/buzz-db/src/runtime/connection_observability.rs @@ -0,0 +1,594 @@ +//! Fixed-vocabulary evidence 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::{ + sync::Arc, + time::{Duration, Instant}, +}; + +/// Database pool roles with connection-lifecycle coverage. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DbPoolRole { + /// The authoritative relay writer pool. + Writer, +} + +impl DbPoolRole { + /// Stable metric/log value. + pub const fn as_str(self) -> &'static str { + match self { + Self::Writer => "writer", + } + } +} + +/// 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 wire 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", + } + } + + /// Stable process-lifecycle phase. + pub const fn lifecycle_phase(self) -> &'static str { + match self { + Self::WriterPool => "db_writer_pool", + Self::PhysicalConnect => "db_physical_connect", + Self::CreatedAtFloor => "db_created_at_floor", + Self::SessionTimeouts => "db_session_timeouts", + Self::Isolation => "db_isolation", + Self::Ready => "db_ready", + } + } +} + +/// Lifecycle edge for one connection-setup step. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DbConnectionEdge { + /// Work on a measurable phase began. + Started, + /// Work reached a bounded terminal. + Terminal, +} + +impl DbConnectionEdge { + /// Stable log value. + pub const fn as_str(self) -> &'static str { + match self { + Self::Started => "started", + Self::Terminal => "terminal", + } + } +} + +/// 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/log value. + pub const fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "succeeded", + Self::Failed => "failed", + Self::TimedOut => "timed_out", + Self::Cancelled => "cancelled", + } + } +} + +/// Secret-safe reason for a failed connection-setup step. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum DbConnectionReason { + /// A generic network I/O error; the SQLx seam cannot safely narrow it. + ConnectIo, + /// TLS negotiation or certificate validation failed. + Tls, + /// PostgreSQL rejected authentication. + Authentication, + /// PostgreSQL rejected the connection for another bounded reason. + ServerReject, + /// A required session-setting query failed. + SessionSetup, + /// The effective transaction isolation was not READ COMMITTED. + IsolationMismatch, + /// SQLx exhausted the pool/connect deadline. + Timeout, + /// The pool was closed. + PoolClosed, + /// A phase guard was dropped with no explicit terminal. + OwnerDropped, + /// A panic unwound through a phase. + Panic, + /// No narrower safe classification exists. + Unknown, +} + +impl DbConnectionReason { + /// Stable log value. + pub const fn as_str(self) -> &'static str { + match self { + Self::ConnectIo => "connect_io", + Self::Tls => "tls", + Self::Authentication => "authentication", + Self::ServerReject => "server_reject", + Self::SessionSetup => "session_setup", + Self::IsolationMismatch => "isolation_mismatch", + Self::Timeout => "timeout", + Self::PoolClosed => "pool_closed", + Self::OwnerDropped => "owner_dropped", + Self::Panic => "panic", + Self::Unknown => "unknown", + } + } +} + +/// One fixed-schema connection lifecycle event. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct DbConnectionLifecycleEvent { + pool_role: DbPoolRole, + connection_ordinal: Option, + step: DbConnectionStep, + edge: DbConnectionEdge, + outcome: Option, + reason: Option, + elapsed: Option, +} + +impl DbConnectionLifecycleEvent { + /// Pool role. + pub const fn pool_role(self) -> DbPoolRole { + self.pool_role + } + + /// Process-local connection ordinal, absent for the aggregate pool phase. + pub const fn connection_ordinal(self) -> Option { + self.connection_ordinal + } + + /// Connection-setup step. + pub const fn step(self) -> DbConnectionStep { + self.step + } + + /// Lifecycle edge. + pub const fn edge(self) -> DbConnectionEdge { + self.edge + } + + /// Terminal outcome, absent from start edges. + pub const fn outcome(self) -> Option { + self.outcome + } + + /// Secret-safe terminal reason. + pub const fn reason(self) -> Option { + self.reason + } + + /// Phase duration, absent from the physical-connect and ready milestones. + pub const fn elapsed(self) -> Option { + self.elapsed + } +} + +/// Sink for fixed-schema database connection lifecycle events. +pub trait DbConnectionObserver: Send + Sync { + /// Record one event. + fn record(&self, event: DbConnectionLifecycleEvent); +} + +#[derive(Default)] +pub(crate) struct NoopDbConnectionObserver; + +impl DbConnectionObserver for NoopDbConnectionObserver { + fn record(&self, _event: DbConnectionLifecycleEvent) {} +} + +pub(crate) type SharedDbConnectionObserver = Arc; + +/// 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 { + observer: SharedDbConnectionObserver, + pool_role: DbPoolRole, + connection_ordinal: Option, + step: DbConnectionStep, + started: Instant, + finished: bool, +} + +impl DbConnectionStepAttempt { + pub(crate) fn start( + observer: SharedDbConnectionObserver, + pool_role: DbPoolRole, + connection_ordinal: Option, + step: DbConnectionStep, + ) -> Self { + metrics::counter!( + "buzz_db_connection_step_started_total", + "pool_role" => pool_role.as_str(), + "step" => step.as_str(), + ) + .increment(1); + observer.record(DbConnectionLifecycleEvent { + pool_role, + connection_ordinal, + step, + edge: DbConnectionEdge::Started, + outcome: None, + reason: None, + elapsed: None, + }); + Self { + observer, + pool_role, + connection_ordinal, + step, + started: Instant::now(), + finished: false, + } + } + + pub(crate) fn succeed(self) { + self.finish(DbConnectionOutcome::Succeeded, None); + } + + pub(crate) fn fail(self, reason: DbConnectionReason) { + self.finish(DbConnectionOutcome::Failed, Some(reason)); + } + + pub(crate) fn time_out(self) { + self.finish( + DbConnectionOutcome::TimedOut, + Some(DbConnectionReason::Timeout), + ); + } + + fn finish(mut self, outcome: DbConnectionOutcome, reason: Option) { + let elapsed = self.started.elapsed(); + record_terminal( + &self.observer, + self.pool_role, + self.connection_ordinal, + self.step, + outcome, + reason, + Some(elapsed), + ); + self.finished = true; + } +} + +impl Drop for DbConnectionStepAttempt { + fn drop(&mut self) { + if self.finished { + return; + } + let (outcome, reason) = if std::thread::panicking() { + (DbConnectionOutcome::Failed, DbConnectionReason::Panic) + } else { + ( + DbConnectionOutcome::Cancelled, + DbConnectionReason::OwnerDropped, + ) + }; + record_terminal( + &self.observer, + self.pool_role, + self.connection_ordinal, + self.step, + outcome, + Some(reason), + Some(self.started.elapsed()), + ); + self.finished = true; + } +} + +pub(crate) fn record_milestone( + observer: &SharedDbConnectionObserver, + pool_role: DbPoolRole, + connection_ordinal: u64, + step: DbConnectionStep, +) { + record_terminal( + observer, + pool_role, + Some(connection_ordinal), + step, + DbConnectionOutcome::Succeeded, + None, + None, + ); +} + +fn record_terminal( + observer: &SharedDbConnectionObserver, + pool_role: DbPoolRole, + connection_ordinal: Option, + step: DbConnectionStep, + outcome: DbConnectionOutcome, + reason: Option, + 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()); + } + observer.record(DbConnectionLifecycleEvent { + pool_role, + connection_ordinal, + step, + edge: DbConnectionEdge::Terminal, + outcome: Some(outcome), + reason, + elapsed, + }); +} + +pub(crate) fn classify_pool_error( + error: &sqlx::Error, +) -> (DbConnectionOutcome, DbConnectionReason) { + match error { + sqlx::Error::PoolTimedOut => (DbConnectionOutcome::TimedOut, DbConnectionReason::Timeout), + sqlx::Error::PoolClosed => (DbConnectionOutcome::Failed, DbConnectionReason::PoolClosed), + sqlx::Error::Io(_) => (DbConnectionOutcome::Failed, DbConnectionReason::ConnectIo), + sqlx::Error::Tls(_) => (DbConnectionOutcome::Failed, DbConnectionReason::Tls), + sqlx::Error::Database(error) + if matches!(error.code().as_deref(), Some("28P01" | "28000")) => + { + ( + DbConnectionOutcome::Failed, + DbConnectionReason::Authentication, + ) + } + sqlx::Error::Database(_) => ( + DbConnectionOutcome::Failed, + DbConnectionReason::ServerReject, + ), + _ => (DbConnectionOutcome::Failed, DbConnectionReason::Unknown), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use std::sync::Mutex; + + #[derive(Default)] + struct CapturingObserver(Mutex>); + + impl DbConnectionObserver for CapturingObserver { + fn record(&self, event: DbConnectionLifecycleEvent) { + self.0.lock().expect("capture DB event").push(event); + } + } + + #[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 observer = Arc::new(CapturingObserver::default()); + let attempt = DbConnectionStepAttempt::start( + observer.clone(), + DbPoolRole::Writer, + Some(7), + DbConnectionStep::CreatedAtFloor, + ); + drop(attempt); + + let events = observer.0.lock().expect("read DB events").clone(); + assert_eq!(events.len(), 2); + assert_eq!(events[0].edge(), DbConnectionEdge::Started); + assert_eq!(events[1].edge(), DbConnectionEdge::Terminal); + assert_eq!(events[1].outcome(), Some(DbConnectionOutcome::Cancelled)); + assert_eq!(events[1].reason(), Some(DbConnectionReason::OwnerDropped)); + + 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), + DebugValue::Histogram(values) => assert_eq!(values.len(), 1), + DebugValue::Gauge(_) => panic!("connection lifecycle has no gauges"), + } + } + } + + #[test] + fn pool_errors_use_only_bounded_classes() { + assert_eq!( + classify_pool_error(&sqlx::Error::PoolTimedOut), + (DbConnectionOutcome::TimedOut, DbConnectionReason::Timeout) + ); + assert_eq!( + classify_pool_error(&sqlx::Error::PoolClosed), + (DbConnectionOutcome::Failed, DbConnectionReason::PoolClosed) + ); + 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_error(&io), + (DbConnectionOutcome::Failed, DbConnectionReason::ConnectIo) + ); + } +} diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 5f608cb78fc..ce5f3c985c2 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -1,13 +1,29 @@ +mod connection_observability; pub mod migration; pub(crate) mod observability; pub mod replica_fence; +pub use connection_observability::{ + DbConnectionEdge, DbConnectionLifecycleEvent, DbConnectionObserver, DbConnectionOutcome, + DbConnectionReason, DbConnectionStep, DbPoolRole, +}; +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}; use sqlx::postgres::PgPoolOptions; use sqlx::{PgPool, QueryBuilder}; -use std::time::Duration; +use std::{ + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + }, + time::Duration, +}; use uuid::Uuid; use buzz_core::{CommunityId, StoredEvent}; @@ -555,7 +571,23 @@ impl Db { /// `buzz.created_at_floor` GUC — this is what makes the replica fence /// proof hold for every insert path that goes through this pool. pub async fn new(config: &DbConfig) -> Result { - let pool = Self::connect_writer_pool(config).await?; + Self::new_with_connection_observer( + config, + Arc::new(connection_observability::NoopDbConnectionObserver), + ) + .await + } + + /// Create a `Db` while publishing writer connection lifecycle events. + /// + /// The observer receives only fixed enums, durations, and process-local + /// ordinals. Database URLs, hosts, usernames, SQL, and raw errors never + /// cross this boundary. + pub async fn new_with_connection_observer( + config: &DbConfig, + observer: Arc, + ) -> Result { + let pool = Self::connect_writer_pool_with_observer(config, observer).await?; let read_max_connections = config .read_max_connections .unwrap_or(config.max_connections); @@ -584,9 +616,28 @@ 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 { + Self::connect_writer_pool_with_observer( + config, + Arc::new(connection_observability::NoopDbConnectionObserver), + ) + .await + } + + /// Connect the writer pool and publish fixed-schema connection events. + pub async fn connect_writer_pool_with_observer( + config: &DbConfig, + observer: Arc, + ) -> Result { + use connection_observability::{ + classify_pool_error, record_milestone, DbConnectionReason, DbConnectionStep, + DbConnectionStepAttempt, DbPoolRole, + }; + 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; + let next_connection_ordinal = Arc::new(AtomicU64::new(1)); + let hook_observer = Arc::clone(&observer); let options = PgPoolOptions::new() .max_connections(config.max_connections) .min_connections(config.min_connections) @@ -594,12 +645,37 @@ impl Db { .max_lifetime(Duration::from_secs(config.max_lifetime_secs)) .idle_timeout(Duration::from_secs(config.idle_timeout_secs)) .after_connect(move |conn, _meta| { + let observer = Arc::clone(&hook_observer); + let connection_ordinal = next_connection_ordinal.fetch_add(1, Ordering::Relaxed); 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( + &observer, + DbPoolRole::Writer, + connection_ordinal, + 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( + Arc::clone(&observer), + DbPoolRole::Writer, + Some(connection_ordinal), + 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(DbConnectionReason::SessionSetup); + 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, @@ -608,7 +684,13 @@ 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( + Arc::clone(&observer), + DbPoolRole::Writer, + Some(connection_ordinal), + 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)", @@ -617,11 +699,31 @@ 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(DbConnectionReason::SessionSetup); + return Err(error); + } + timeouts.succeed(); + + let isolation_step = DbConnectionStepAttempt::start( + Arc::clone(&observer), + DbPoolRole::Writer, + Some(connection_ordinal), + 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(DbConnectionReason::SessionSetup); + return Err(error); + } + }; if isolation != "read committed" { + isolation_step.fail(DbConnectionReason::IsolationMismatch); return Err(sqlx::Error::Configuration( format!( "writer pool requires READ COMMITTED transaction isolation, got {isolation}" @@ -629,10 +731,38 @@ impl Db { .into(), )); } + isolation_step.succeed(); + record_milestone( + &observer, + DbPoolRole::Writer, + connection_ordinal, + DbConnectionStep::Ready, + ); Ok(()) }) }); - Ok(options.connect(&config.database_url).await?) + + let pool_attempt = DbConnectionStepAttempt::start( + Arc::clone(&observer), + DbPoolRole::Writer, + None, + DbConnectionStep::WriterPool, + ); + match options.connect(&config.database_url).await { + Ok(pool) => { + pool_attempt.succeed(); + Ok(pool) + } + Err(error) => { + let (outcome, reason) = classify_pool_error(&error); + if outcome == connection_observability::DbConnectionOutcome::TimedOut { + pool_attempt.time_out(); + } else { + pool_attempt.fail(reason); + } + 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 4d2e5f7ad69..e1400e02f29 100644 --- a/crates/buzz-db/src/runtime/observability.rs +++ b/crates/buzz-db/src/runtime/observability.rs @@ -151,8 +151,8 @@ pub(crate) const POOL_ACQUIRE_VALID_PAIRS: [(&str, &str); 11] = [ ("writer", "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 { @@ -380,6 +380,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() @@ -604,7 +610,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), [ @@ -861,6 +867,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..01b979ea7d6 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2,10 +2,32 @@ use super::*; use crate::{relay_members, thread}; use buzz_core::CommunityId; use sqlx::{Connection, PgPool}; +use std::sync::{Arc, Mutex}; use uuid::Uuid; const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials +#[derive(Default)] +struct CapturingConnectionObserver(Mutex>); + +impl DbConnectionObserver for CapturingConnectionObserver { + fn record(&self, event: DbConnectionLifecycleEvent) { + self.0 + .lock() + .expect("capture connection lifecycle event") + .push(event); + } +} + +impl CapturingConnectionObserver { + fn events(&self) -> Vec { + self.0 + .lock() + .expect("read connection lifecycle events") + .clone() + } +} + 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) @@ -2516,7 +2538,7 @@ async fn created_at_floor_guard_aborts_old_channel_rows_at_commit() { fn writer_pool_safety_hook_is_single_and_composed() { let source = include_str!("mod.rs"); let connect_pool = source - .split("async fn connect_writer_pool") + .split("async fn connect_writer_pool_with_observer") .nth(1) .and_then(|tail| tail.split("const READER_ACQUIRE_TIMEOUT").next()) .expect("connect_writer_pool source block"); @@ -2544,6 +2566,122 @@ fn writer_pool_safety_hook_is_single_and_composed() { assert!(!reader_doc.contains("Db::connect_writer_pool")); } +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn writer_pool_observer_records_every_initial_connection_ready() { + let observer = Arc::new(CapturingConnectionObserver::default()); + let db = Db::new_with_connection_observer( + &DbConfig { + database_url: crate::test_support::database_url(), + max_connections: 2, + min_connections: 2, + ..DbConfig::default() + }, + observer.clone(), + ) + .await + .expect("connect observed writer pool"); + + let events = observer.events(); + assert_eq!( + events.first().map(|event| (event.step(), event.edge())), + Some((DbConnectionStep::WriterPool, DbConnectionEdge::Started)) + ); + assert_eq!( + events.last().map(|event| (event.step(), event.outcome())), + Some(( + DbConnectionStep::WriterPool, + Some(DbConnectionOutcome::Succeeded) + )) + ); + for ordinal in [1, 2] { + let connection_events = events + .iter() + .filter(|event| event.connection_ordinal() == Some(ordinal)) + .collect::>(); + assert_eq!( + connection_events + .iter() + .filter(|event| event.edge() == DbConnectionEdge::Terminal) + .map(|event| (event.step(), event.outcome())) + .collect::>(), + [ + ( + DbConnectionStep::PhysicalConnect, + Some(DbConnectionOutcome::Succeeded) + ), + ( + DbConnectionStep::CreatedAtFloor, + Some(DbConnectionOutcome::Succeeded) + ), + ( + DbConnectionStep::SessionTimeouts, + Some(DbConnectionOutcome::Succeeded) + ), + ( + DbConnectionStep::Isolation, + Some(DbConnectionOutcome::Succeeded) + ), + ( + DbConnectionStep::Ready, + Some(DbConnectionOutcome::Succeeded) + ), + ] + ); + } + db.pool.close().await; +} + +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn writer_pool_observer_records_connection_created_after_startup() { + let observer = Arc::new(CapturingConnectionObserver::default()); + let pool = Db::connect_writer_pool_with_observer( + &DbConfig { + database_url: crate::test_support::database_url(), + max_connections: 2, + min_connections: 1, + ..DbConfig::default() + }, + observer.clone(), + ) + .await + .expect("connect observed size-one writer pool"); + let initial_ready = observer + .events() + .iter() + .filter(|event| event.step() == DbConnectionStep::Ready) + .count(); + assert_eq!(initial_ready, 1); + + let first = pool.acquire().await.expect("hold initial connection"); + let second = pool + .acquire() + .await + .expect("grow pool with a second connection"); + let events = observer.events(); + assert_eq!( + events + .iter() + .filter(|event| { + event.step() == DbConnectionStep::Ready + && event.outcome() == Some(DbConnectionOutcome::Succeeded) + }) + .count(), + 2, + "a post-startup pool growth connection must traverse the same observed safety hook" + ); + assert!(events.iter().any(|event| { + event.connection_ordinal() == Some(2) + && event.step() == DbConnectionStep::PhysicalConnect + && event.outcome() == Some(DbConnectionOutcome::Succeeded) + })); + + drop(second); + drop(first); + pool.close().await; +} + #[tokio::test] #[ignore = "requires Postgres"] async fn writer_pool_rejects_non_read_committed_database_default() { @@ -2562,13 +2700,17 @@ async fn writer_pool_rejects_non_read_committed_database_default() { let base = admin_url().await; let idx = base.rfind('/').expect("db url has a path segment"); let scratch_url = format!("{}/{}", &base[..idx], name); - let error = Db::new(&DbConfig { - database_url: scratch_url, - max_connections: 1, - min_connections: 1, - acquire_timeout_secs: 1, - ..DbConfig::default() - }) + let observer = Arc::new(CapturingConnectionObserver::default()); + let error = Db::new_with_connection_observer( + &DbConfig { + database_url: scratch_url, + max_connections: 1, + min_connections: 1, + acquire_timeout_secs: 1, + ..DbConfig::default() + }, + observer.clone(), + ) .await .expect_err("writer pool must reject pinned-snapshot database defaults"); assert!( @@ -2576,6 +2718,18 @@ async fn writer_pool_rejects_non_read_committed_database_default() { || error.to_string().contains("pool timed out"), "unexpected isolation rejection: {error}" ); + let events = observer.events(); + assert!(events.iter().any(|event| { + event.step() == DbConnectionStep::Isolation + && event.outcome() == Some(DbConnectionOutcome::Failed) + && event.reason() == Some(DbConnectionReason::IsolationMismatch) + })); + assert!( + !events + .iter() + .any(|event| event.step() == DbConnectionStep::Ready), + "an isolation-rejected connection must never emit ready" + ); sqlx::query(sqlx::AssertSqlSafe(format!( "DROP DATABASE {name} WITH (FORCE)" @@ -2585,6 +2739,40 @@ async fn writer_pool_rejects_non_read_committed_database_default() { .expect("drop isolation test database"); } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn writer_pool_observer_stops_after_session_timeout_setup_failure() { + let observer = Arc::new(CapturingConnectionObserver::default()); + let error = Db::new_with_connection_observer( + &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() + }, + observer.clone(), + ) + .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 events = observer.events(); + assert!(events.iter().any(|event| { + event.step() == DbConnectionStep::SessionTimeouts + && event.outcome() == Some(DbConnectionOutcome::Failed) + && event.reason() == Some(DbConnectionReason::SessionSetup) + })); + assert!(!events.iter().any(|event| matches!( + event.step(), + DbConnectionStep::Isolation | DbConnectionStep::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..fb92fb053b8 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() diff --git a/crates/buzz-relay/src/lifecycle.rs b/crates/buzz-relay/src/lifecycle.rs index bc9d51062b2..d142ad8e0e8 100644 --- a/crates/buzz-relay/src/lifecycle.rs +++ b/crates/buzz-relay/src/lifecycle.rs @@ -1,4 +1,4 @@ -//! Fixed-schema evidence for the relay's earliest startup steps. +//! Fixed-schema evidence for relay startup and database connection setup. //! //! These events are written directly to stderr because crypto, tracing, //! configuration, and metrics setup can fail before the normal telemetry @@ -17,6 +17,11 @@ use std::{ use serde::Serialize; use uuid::Uuid; +use buzz_db::{ + DbConnectionEdge, DbConnectionLifecycleEvent, DbConnectionObserver, DbConnectionOutcome, + DbConnectionReason, DbConnectionStep, DbPoolRole, +}; + const EVENT_NAME: &str = "buzz_process_lifecycle"; const SCHEMA_VERSION: u8 = 1; @@ -168,6 +173,10 @@ struct LifecycleEvent { process_elapsed_ms: u64, #[serde(skip_serializing_if = "Option::is_none")] phase_elapsed_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pool_role: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + connection_ordinal: Option, } trait EventWriter: Send + Sync { @@ -214,7 +223,7 @@ impl ProcessLifecycle { } else { Instant::now() }; - self.emit(phase, "started", None, None, None); + self.emit_startup(phase, "started", None, None, None); PhaseGuard { lifecycle: Arc::clone(self), phase, @@ -223,28 +232,54 @@ impl ProcessLifecycle { } } - fn emit( + fn emit_startup( &self, phase: StartupPhase, edge: &'static str, status: Option, reason: Option, elapsed: Option, + ) { + self.emit( + "startup", + phase.as_str(), + edge, + status.map(LifecycleStatus::as_str), + reason.map(LifecycleReason::as_str), + elapsed, + None, + None, + ); + } + + #[allow(clippy::too_many_arguments)] + fn emit( + &self, + track: &'static str, + phase: &'static str, + edge: &'static str, + status: Option<&'static str>, + reason: Option<&'static str>, + elapsed: Option, + pool_role: Option<&'static str>, + connection_ordinal: Option, ) { self.writer.emit(&LifecycleEvent { event_name: EVENT_NAME, schema_version: SCHEMA_VERSION, process_boot_id: self.boot_id, sequence: self.sequence.fetch_add(1, Ordering::Relaxed), - track: "startup", - phase: phase.as_str(), + track, + phase, edge, - status: status.map(LifecycleStatus::as_str), - reason: reason.map(LifecycleReason::as_str), + status, + reason, process_started_at_unix_ms: millis_since_epoch(self.wall_origin), observed_at_unix_ms: millis_since_epoch(SystemTime::now()), process_elapsed_ms: saturating_millis(self.monotonic_origin.elapsed()), phase_elapsed_ms: elapsed.map(saturating_millis), + pool_role, + connection_ordinal, }); } } @@ -276,7 +311,7 @@ impl PhaseGuard { fn finish(mut self, status: LifecycleStatus, reason: Option) { let elapsed = self.started_at.elapsed(); self.lifecycle - .emit(self.phase, "terminal", Some(status), reason, Some(elapsed)); + .emit_startup(self.phase, "terminal", Some(status), reason, Some(elapsed)); self.finished = true; } } @@ -291,7 +326,7 @@ impl Drop for PhaseGuard { } else { (LifecycleStatus::Abandoned, LifecycleReason::OwnerDropped) }; - self.lifecycle.emit( + self.lifecycle.emit_startup( self.phase, "terminal", Some(status), @@ -371,13 +406,15 @@ impl BootTracker { } /// Finish early startup with a structured lifecycle terminal. - pub fn finish(self) { + pub fn finish(self) -> LifecycleRecorder { let status = if self.degraded.is_some() { LifecycleStatus::Degraded } else { LifecycleStatus::Succeeded }; + let lifecycle = Arc::clone(&self.lifecycle); self.headline.finish(status, self.degraded); + LifecycleRecorder { lifecycle } } fn fail(self, reason: LifecycleReason) { @@ -385,6 +422,51 @@ impl BootTracker { } } +/// Cloneable bridge from dependency lifecycle events into the process schema. +#[derive(Clone)] +pub struct LifecycleRecorder { + lifecycle: Arc, +} + +impl LifecycleRecorder { + #[allow(clippy::too_many_arguments)] + fn record_database_fields( + &self, + pool_role: DbPoolRole, + connection_ordinal: Option, + step: DbConnectionStep, + edge: DbConnectionEdge, + outcome: Option, + reason: Option, + elapsed: Option, + ) { + self.lifecycle.emit( + "database", + step.lifecycle_phase(), + edge.as_str(), + outcome.map(|outcome| outcome.as_str()), + reason.map(|reason| reason.as_str()), + elapsed, + Some(pool_role.as_str()), + connection_ordinal, + ); + } +} + +impl DbConnectionObserver for LifecycleRecorder { + fn record(&self, event: DbConnectionLifecycleEvent) { + self.record_database_fields( + event.pool_role(), + event.connection_ordinal(), + event.step(), + event.edge(), + event.outcome(), + event.reason(), + event.elapsed(), + ); + } +} + fn millis_since_epoch(time: SystemTime) -> u64 { time.duration_since(UNIX_EPOCH) .map(saturating_millis) @@ -588,4 +670,39 @@ mod tests { ] ); } + + #[test] + fn database_events_share_boot_sequence_without_sensitive_fields() { + let (lifecycle, writer) = recorder(); + let recorder = LifecycleRecorder { lifecycle }; + recorder.record_database_fields( + DbPoolRole::Writer, + Some(2), + DbConnectionStep::Isolation, + DbConnectionEdge::Terminal, + Some(DbConnectionOutcome::Failed), + Some(DbConnectionReason::IsolationMismatch), + Some(Duration::from_millis(4)), + ); + + let values = events(&writer); + assert_eq!(values.len(), 1); + let value = serde_json::to_value(&values[0]).expect("serialize database lifecycle event"); + assert_eq!(value["sequence"], 1); + assert_eq!(value["track"], "database"); + assert_eq!(value["phase"], "db_isolation"); + assert_eq!(value["edge"], "terminal"); + assert_eq!(value["status"], "failed"); + assert_eq!(value["reason"], "isolation_mismatch"); + assert_eq!(value["pool_role"], "writer"); + assert_eq!(value["connection_ordinal"], 2); + assert_eq!(value["phase_elapsed_ms"], 4); + let rendered = serde_json::to_string(&value).expect("render database lifecycle event"); + for forbidden in ["postgres://", "password", "database_url", "hostname", "sql"] { + assert!( + !rendered.contains(forbidden), + "leaked forbidden field {forbidden}" + ); + } + } } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 206f0329c0e..578cee8d298 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -238,7 +238,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { relay_metrics::MetricsInstallFailure::ExporterBuild => LifecycleReason::ExporterBuild, }, )?; - boot.finish(); + let lifecycle = boot.finish(); metrics::gauge!("buzz_audit_enabled").set(if config.audit_enabled { 1.0 } else { 0.0 }); metrics::gauge!("buzz_push_enabled").set(if config.push_enabled { 1.0 } else { 0.0 }); info!( @@ -256,10 +256,12 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { ..DbConfig::default() } .with_session_timeouts_from_env(); - let db = Db::new(&db_config).await.map_err(|e| { - error!("Failed to connect to Postgres: {e}"); - anyhow::anyhow!("DB connection failed: {e}") - })?; + let db = Db::new_with_connection_observer(&db_config, Arc::new(lifecycle)) + .await + .map_err(|e| { + error!("Failed to connect to Postgres: {e}"); + anyhow::anyhow!("DB connection failed: {e}") + })?; if db.has_read_pool() { info!("Postgres connected (writer + lazy read replica pool)"); // Reader-down at boot must not crash or block the relay; this warn-only diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index f71894116c3..de19027ceab 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" + ); } #[cfg(test)] @@ -345,11 +373,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, @@ -372,15 +406,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() @@ -405,7 +468,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{") }) @@ -420,7 +484,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"])); @@ -430,6 +496,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/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs index 29fcf991f8d..8259adb8c3e 100644 --- a/crates/buzz-relay/tests/boot_lifecycle.rs +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -226,14 +226,19 @@ fn assert_accounting(events: &[Value]) { assert_eq!(event["schema_version"], 1); assert_eq!(event["sequence"], u64::try_from(index + 1).unwrap()); assert_eq!(event["process_boot_id"], boot_id); - assert_eq!(event["track"], "startup"); - let count = counts - .entry(event["phase"].as_str().expect("phase").to_owned()) - .or_default(); - match event["edge"].as_str() { - Some("started") => count.0 += 1, - Some("terminal") => count.1 += 1, - other => panic!("unexpected lifecycle edge: {other:?}"), + match event["track"].as_str() { + Some("startup") => { + let count = counts + .entry(event["phase"].as_str().expect("phase").to_owned()) + .or_default(); + match event["edge"].as_str() { + Some("started") => count.0 += 1, + Some("terminal") => count.1 += 1, + other => panic!("unexpected startup lifecycle edge: {other:?}"), + } + } + Some("database") => assert_database_schema(event), + other => panic!("unexpected lifecycle track: {other:?}"), } } assert!( @@ -244,6 +249,28 @@ fn assert_accounting(events: &[Value]) { ); } +fn assert_database_schema(event: &Value) { + assert!( + [ + "db_writer_pool", + "db_physical_connect", + "db_created_at_floor", + "db_session_timeouts", + "db_isolation", + "db_ready", + ] + .contains(&event["phase"].as_str().expect("database phase")), + "unexpected database phase: {}", + event["phase"], + ); + assert!( + matches!(event["edge"].as_str(), Some("started" | "terminal")), + "unexpected database edge: {}", + event["edge"], + ); + assert_eq!(event["pool_role"], "writer"); +} + fn assert_terminal(events: &[Value], phase: &str, status: &str, reason: Option<&str>) { let terminal = events .iter() diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 5e778279130..034ca4ee0ca 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,42 @@ 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. + +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. Fixed-schema `buzz_process_lifecycle` logs preserve exact per-pod event +order and process-local connection ordinals; metrics provide bucketed trends. + ## Relay Pod extensions The chart exposes narrow extension points for init containers, volumes, relay