diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index de8f6d3d1c..fa3f06961f 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -969,17 +969,48 @@ impl Db { /// the reason names the mechanism rather than a diagnosis). const READER_ACQUIRE_TIMEOUT: Duration = Duration::from_millis(150); + /// Client-side bound on a routed read's fence proof (Aurora capability + /// probe, `BEGIN`, heartbeat observation) once a connection is in hand. + /// Well above a normal round trip, so a brief reader stall waits on the + /// reader instead of flooding the writer; finite, so a reader that stops + /// answering without closing its sockets fails closed to the writer as + /// `reader_proof_timeout`. + const READER_PROOF_TIMEOUT: Duration = Duration::from_secs(1); + + /// Warm floor for a reader pool of `max_connections`: one fifth of the + /// ceiling, at least one. + /// + /// The floor exists so routed reads find an established connection + /// instead of dialing one inside [`Db::READER_ACQUIRE_TIMEOUT`]. A cold + /// dial (TCP+TLS+SCRAM) that outlasts that budget is dropped mid-handshake + /// (the reader logs `could not accept SSL connection: EOF detected`), the + /// read falls back to the writer, and the next read dials again. Under a + /// fleet-wide burst that loop turns brief reader slowness into writer pool + /// exhaustion. The floor is per relay process: size the reader's + /// `max_connections` for the floor times the number of relay processes. + fn reader_min_connections(max_connections: u32) -> u32 { + (max_connections / 5).max(1).min(max_connections) + } + /// Connect the read-replica pool **lazily** — no connection is /// attempted at construction, so a reader that is down at boot cannot /// crash the relay (it starts all-writer with the fence closed and /// recovers when the replica returns). /// - /// `min_connections` is pinned to 0 explicitly: sqlx's lazy pool still - /// spawns an eager background connect task to satisfy a nonzero - /// minimum, which would reintroduce boot-time reader dial attempts (and - /// their log noise) that "lazy" is meant to avoid. With 0, connections - /// are dialed only on first acquire; the ~10-minute reaper never tops - /// the pool back up, which is fine — routed reads re-fill it on demand. + /// The pool keeps a warm floor of [`Db::reader_min_connections`]. With + /// `max_lifetime` and `idle_timeout` both set, sqlx 0.9 does not dial + /// that floor at construction; it re-dials it in the background, outside + /// the 150ms request budget, from its reaper and whenever a connection is + /// closed instead of released. [`Db::spawn_read_pool_boot_ping`] fills it + /// at boot. + /// + /// `test_before_acquire(false)`: the idle-connection ping runs inside the + /// 150ms acquire budget, and a ping that overruns closes a healthy warm + /// connection instead of returning it. The routed read's first statement + /// (`BEGIN ... READ ONLY` in [`Db::proved_reader`]) already proves the + /// connection and fails closed to the writer as `reader_validation_error`, + /// and every released connection still passes the bounded release check + /// in [`Db::read_pool_options`]. /// /// No floor guard or writer-isolation assertion: replica sessions are /// read-only, so the commit-time trigger from migration 0021 never fires @@ -990,12 +1021,65 @@ impl Db { /// the server/role default is already tighter, so no replica read is /// unbounded; per-call `SET LOCAL` caps still apply on top. fn connect_read_pool(config: &DbConfig, url: &str, max_connections: u32) -> Result { - Ok(PgPoolOptions::new() - .max_connections(max_connections) - .min_connections(0) - .acquire_timeout(Self::READER_ACQUIRE_TIMEOUT) + let options: sqlx::postgres::PgConnectOptions = url.parse()?; + Ok(Self::read_pool_options(max_connections) .max_lifetime(Duration::from_secs(config.max_lifetime_secs)) .idle_timeout(Duration::from_secs(config.idle_timeout_secs)) + .connect_lazy_with(Self::reader_connect_options(options))) + } + + /// Reader session options: pins `default_transaction_isolation` to + /// `read committed` in the startup packet (no extra round trip), which + /// overrides any database or role default. The release check in + /// [`Db::read_pool_options`] relies on it: routed reads begin + /// `REPEATABLE READ`, so a session default of `read committed` is what + /// makes a transaction left open by a dropped `BEGIN` visible. + pub(crate) fn reader_connect_options( + options: sqlx::postgres::PgConnectOptions, + ) -> sqlx::postgres::PgConnectOptions { + options.options([("default_transaction_isolation", "read committed")]) + } + + /// Upper bound on a reader connection's release check (see + /// [`Db::read_pool_options`]); matches sqlx's own close-on-drop bound. + const READER_RELEASE_TIMEOUT: Duration = Duration::from_secs(5); + + /// Reader pool options apart from connection lifetimes (see + /// [`Db::connect_read_pool`]). + /// + /// Every checkout that goes back to the pool first passes a release + /// check bounded by [`Db::READER_RELEASE_TIMEOUT`]. sqlx's own release + /// ping has no timeout, so a socket whose reader stopped answering + /// without closing it (a failover, a partition) would otherwise hold its + /// slot until the kernel gives up on it; with every slot held that way, + /// the pool cannot dial the reader that replaced it. A check that times + /// out or errors closes the socket and frees the slot, and sqlx's floor + /// maintenance dials a replacement. + /// + /// The check also closes, rather than returns, a connection still inside + /// a transaction. A checkout dropped while its `BEGIN` was in flight (a + /// timed-out proof, a cancelled caller) has no `ROLLBACK` queued: sqlx + /// counts the transaction only once `BEGIN` is acknowledged. Once the + /// reader answers, that connection would otherwise go back to the pool + /// `idle in transaction`. Closing it ends the session, and the server + /// rolls the transaction back. Routed reads only begin `REPEATABLE READ` + /// (see [`Db::proved_reader`]) and reader sessions default to `READ + /// COMMITTED` (see [`Db::reader_connect_options`]), so an isolation level + /// other than the session default means a transaction is still open. + /// + /// Residual window: after the check passes, sqlx sends its own release + /// ping, which has no timeout. A reader that goes silent in that one + /// round trip strands that one slot until the socket fails. That is + /// bounded by the releases in flight at that instant, and later checkouts + /// of silent connections go through the proof timeout and this check. + /// Checkouts that must never strand a slot use `close_on_drop` (see + /// `ReaderOperation::bounded_release`). + pub(crate) fn read_pool_options(max_connections: u32) -> PgPoolOptions { + PgPoolOptions::new() + .max_connections(max_connections) + .min_connections(Self::reader_min_connections(max_connections)) + .test_before_acquire(false) + .acquire_timeout(Self::READER_ACQUIRE_TIMEOUT) .after_connect(|conn, _meta| { Box::pin(async move { sqlx::query(crate::event::TIGHTEN_STATEMENT_TIMEOUT_SQL) @@ -1006,16 +1090,53 @@ impl Db { .map(|_| ()) }) }) - .connect_lazy(url)?) + .after_release(|conn, _meta| Box::pin(Self::reader_release_check(conn))) } - /// Spawn a one-shot reader reachability probe that only WARNs. + /// The reader pool's release check (see [`Db::read_pool_options`]): + /// `Ok(true)` returns the connection to the pool; anything else closes it + /// and counts the close in `buzz_db_reader_release_closed_total` by + /// `reason`. A rise in `timeout` means reader sockets stopped answering. + async fn reader_release_check( + conn: &mut sqlx::PgConnection, + ) -> std::result::Result { + let outside_transaction = sqlx::query_scalar::<_, bool>( + "SELECT pg_catalog.current_setting('transaction_isolation') \ + = pg_catalog.current_setting('default_transaction_isolation')", + ) + .fetch_one(conn); + let (reason, closed) = + match tokio::time::timeout(Self::READER_RELEASE_TIMEOUT, outside_transaction).await { + Ok(Ok(true)) => return Ok(true), + Ok(Ok(false)) => ("open_transaction", Ok(false)), + Ok(Err(e)) => ("error", Err(e)), + Err(_) => ( + "timeout", + Err(sqlx::Error::Io(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "reader release check timed out", + ))), + ), + }; + metrics::counter!("buzz_db_reader_release_closed_total", "reason" => reason).increment(1); + closed + } + + /// Spawn a one-shot reader reachability probe that only WARNs, then + /// fills the reader's warm floor. /// - /// With a lazy pool and `min_connections(0)`, nothing dials the replica - /// until the first routed read — so a misconfigured `READ_DATABASE_URL` - /// would otherwise be invisible until traffic arrives and quietly falls - /// back to the writer. This ping is the only boot-time reader-down - /// visibility; it must never gate startup or [`Db::spawn_fence_probe`]. + /// The lazy pool dials nothing at construction, so without this ping a + /// misconfigured `READ_DATABASE_URL` would be invisible until traffic + /// arrives and quietly falls back to the writer. This ping is the only + /// boot-time reader-down visibility; it must never gate startup or + /// [`Db::spawn_fence_probe`]. + /// + /// On success it closes its probe connection, which makes sqlx dial the + /// pool up to [`Db::reader_min_connections`] in the background, so a + /// freshly started pod does not meet its first traffic with an empty + /// reader pool and pay every dial inside the 150ms request budget. + /// Probe acquires that overrun that budget are retried with + /// [`Db::READER_BOOT_PROBE_BACKOFF`]. /// /// On success it also primes the Aurora identity capability cache /// ([`Db::reader_aurora_identity`]) on the connection it already holds, @@ -1032,32 +1153,58 @@ impl Db { tokio::spawn(Self::read_pool_boot_ping_once(read_pool, aurora_identity)); } + /// Backoff before each boot-probe retry after an acquire that overran + /// [`Db::READER_ACQUIRE_TIMEOUT`]. Bounded and spaced out so a reader + /// that is slow at boot sees a handful of abandoned dials per pod, not + /// a tight retry loop; off the request path, so a slow boot dial costs + /// a retry, not a fallback. + const READER_BOOT_PROBE_BACKOFF: [Duration; 4] = [ + Duration::from_millis(250), + Duration::from_millis(500), + Duration::from_secs(1), + Duration::from_secs(2), + ]; + async fn read_pool_boot_ping_once( read_pool: PgPool, aurora_identity: std::sync::Arc>, ) { - match observability::acquire_reader_with_legacy_metrics( - &read_pool, - observability::ReaderOperation::Bootstrap, - ) - .await - { - Ok(mut conn) => { - tracing::info!("read replica reachable at boot"); - match replica_fence::reader_supports_aurora_identity(&mut conn).await { - Ok(supported) => { - let _ = aurora_identity.set(supported); - } - Err(e) => tracing::debug!( - error = %e, - "aurora identity boot prime failed; first routed read will probe" - ), + let mut backoff = Self::READER_BOOT_PROBE_BACKOFF.into_iter(); + let mut conn = loop { + let error = match observability::acquire_reader_with_legacy_metrics( + &read_pool, + observability::ReaderOperation::Bootstrap, + ) + .await + { + Ok(conn) => break conn, + Err(error) => error, + }; + match (error, backoff.next()) { + (sqlx::Error::PoolTimedOut, Some(delay)) => tokio::time::sleep(delay).await, + (e, _) => { + tracing::warn!( + "read replica unreachable at boot; serving all-writer until it recovers: {e}" + ); + return; } } - Err(e) => tracing::warn!( - "read replica unreachable at boot; serving all-writer until it recovers: {e}" + }; + tracing::info!("read replica reachable at boot"); + match replica_fence::reader_supports_aurora_identity(&mut conn).await { + Ok(supported) => { + let _ = aurora_identity.set(supported); + } + Err(e) => tracing::debug!( + error = %e, + "aurora identity boot prime failed; first routed read will probe" ), } + // Closing (rather than returning) the probe connection makes sqlx + // run its `min_connections` maintenance, which dials the warm floor + // in the background under its own 300s deadline, outside the 150ms + // request budget, releasing each connection idle as it lands. + conn.close_on_drop(); } #[cfg(test)] @@ -1226,54 +1373,70 @@ impl Db { // can drop this checkout at any of them. conn.close_on_drop(); } - let aurora = self.reader_aurora_capability_on(&mut conn).await; - let mut tx = match sqlx::Transaction::begin( - conn, - Some(sqlx::SqlStr::from_static( - "BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY", - )), - ) - .await - { - Ok(tx) => tx, - // The acquire miss gets its own reason code: the reader pool's - // short acquire timeout (READER_ACQUIRE_TIMEOUT) makes this the - // fast fail-closed path under load, and - // `buzz_db_route_decision{decision="writer",reason="reader_acquire_timeout"}` - // is the operator's alert signal for a struggling reader pool. - // - // The reason deliberately names the mechanism, not a diagnosis: - // `PoolTimedOut` proves only that no connection was handed out - // within the 150ms budget. That budget includes cold connect - // (TCP+TLS+auth), and sqlx's `size` counts in-flight dials, so - // this fires for slow connection establishment as well as for - // established-connection contention — and neither `size == 0` - // nor `size >= max` recovers the missing causal bit (in-flight - // dials hold a size slot, and a cold burst can push - // `active = size - idle` toward max with zero busy connections). - // Runbook: correlate with `buzz_db_read_pool_active` / `_max` - // and reader connection health/latency; high active suggests - // contention, but this metric alone does not distinguish - // contention from slow connects. Note the gauge is a coarse - // sample (BUZZ_POOL_METRICS_INTERVAL_SECS, default 10s) while - // the event it explains lasts ~150ms — a short burst may fall - // between samples entirely, so absence of elevated active is - // NOT evidence of a cold connect. - Err(sqlx::Error::PoolTimedOut) => { - tracing::warn!("reader pool acquire timed out"); - return Err("reader_acquire_timeout"); - } - Err(e) => { - tracing::warn!(error = %e, "reader transaction begin failed"); - return Err("reader_validation_error"); - } + // The proof is the reader's first wire round trip (no pre-acquire + // ping, see `connect_read_pool`), so it carries the liveness bound: + // a half-open socket to a vanished reader would otherwise hang the + // request until TCP gives up. An expired proof is simply dropped: + // the reader pool's release check (`read_pool_options`) bounds how + // long its connection holds a slot and closes it if `BEGIN` landed. + let proof = async move { + let aurora = self.reader_aurora_capability_on(&mut conn).await; + let mut tx = match sqlx::Transaction::begin( + conn, + Some(sqlx::SqlStr::from_static( + "BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY", + )), + ) + .await + { + Ok(tx) => tx, + // The acquire miss gets its own reason code: the reader pool's + // short acquire timeout (READER_ACQUIRE_TIMEOUT) makes this the + // fast fail-closed path under load, and + // `buzz_db_route_decision{decision="writer",reason="reader_acquire_timeout"}` + // is the operator's alert signal for a struggling reader pool. + // + // The reason deliberately names the mechanism, not a diagnosis: + // `PoolTimedOut` proves only that no connection was handed out + // within the 150ms budget. That budget includes cold connect + // (TCP+TLS+auth), and sqlx's `size` counts in-flight dials, so + // this fires for slow connection establishment as well as for + // established-connection contention — and neither `size == 0` + // nor `size >= max` recovers the missing causal bit (in-flight + // dials hold a size slot, and a cold burst can push + // `active = size - idle` toward max with zero busy connections). + // Runbook: correlate with `buzz_db_read_pool_active` / `_max` + // and reader connection health/latency; high active suggests + // contention, but this metric alone does not distinguish + // contention from slow connects. Note the gauge is a coarse + // sample (BUZZ_POOL_METRICS_INTERVAL_SECS, default 10s) while + // the event it explains lasts ~150ms — a short burst may fall + // between samples entirely, so absence of elevated active is + // NOT evidence of a cold connect. + Err(sqlx::Error::PoolTimedOut) => { + tracing::warn!("reader pool acquire timed out"); + return Err("reader_acquire_timeout"); + } + Err(e) => { + tracing::warn!(error = %e, "reader transaction begin failed"); + return Err("reader_validation_error"); + } + }; + let obs = match replica_fence::observe_heartbeat(&mut tx, aurora).await { + Ok(Some(observation)) => observation, + Ok(None) => return Err("reader_validation_error"), + Err(e) => { + tracing::warn!(error = %e, "heartbeat observation failed"); + return Err("reader_validation_error"); + } + }; + Ok((tx, obs)) }; - let obs = match replica_fence::observe_heartbeat(&mut tx, aurora).await { - Ok(Some(observation)) => observation, - Ok(None) => return Err("reader_validation_error"), - Err(e) => { - tracing::warn!(error = %e, "heartbeat observation failed"); - return Err("reader_validation_error"); + let (tx, obs) = match tokio::time::timeout(Self::READER_PROOF_TIMEOUT, proof).await { + Ok(proved) => proved?, + Err(_) => { + tracing::warn!("reader fence proof timed out"); + return Err("reader_proof_timeout"); } }; match self.fence.resolve(obs.token, obs.epoch) { diff --git a/crates/buzz-db/src/runtime/observability.rs b/crates/buzz-db/src/runtime/observability.rs index 7220896db4..6117eb7835 100644 --- a/crates/buzz-db/src/runtime/observability.rs +++ b/crates/buzz-db/src/runtime/observability.rs @@ -103,11 +103,11 @@ impl ReaderOperation { /// bounded `close_on_drop` instead of returning it to the pool. /// /// Maintenance reads run under relay-side deadlines that drop them - /// mid-query when a reader goes dark. SQLx's return-to-pool path pings - /// that connection with no timeout while still holding a pool slot, and - /// the reader pool is shared with serving reads, so each abandoned read - /// would strand a serving slot until the kernel gives up on the socket. - /// Serving reads keep pooled reuse. + /// mid-query when a reader goes dark, and the reader pool is shared with + /// serving reads. The pool's release check is bounded, but SQLx's own + /// release ping after it is not, so a returned connection whose reader + /// goes silent in that window still strands its slot. Close-on-drop + /// bounds the whole release. Serving reads keep pooled reuse. pub(crate) const fn bounded_release(self) -> bool { matches!(self, Self::Maintenance) } diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 85e9722280..c9e1c6dc36 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -1139,30 +1139,52 @@ async fn read_pool_stats_reports_reader_ceiling_not_writer() { ); } -/// D4 wiring: the reader pool is built lazily with `min_connections(0)` -/// and the short reader acquire timeout — construction must succeed -/// with no replica listening (reader-down at boot must not crash the -/// relay), and `read_max_connections` must honour -/// `DbConfig::read_max_connections` over the writer sizing. +/// D4 wiring: the reader pool is built lazily with the short reader +/// acquire timeout, a warm `min_connections` floor, and no pre-acquire +/// ping — construction must succeed with no replica listening +/// (reader-down at boot must not crash the relay), and +/// `read_max_connections` must honour `DbConfig::read_max_connections` +/// over the writer sizing. /// `#[tokio::test]` because `connect_lazy` spawns the pool reaper task, /// which needs a Tokio runtime even though nothing is dialed. #[tokio::test] async fn connect_read_pool_is_lazy_and_independently_sized() { let config = DbConfig { max_connections: 20, - read_max_connections: Some(7), + read_max_connections: Some(10), ..DbConfig::default() }; // Unroutable per RFC 5737 TEST-NET-1: proves nothing is dialed at // construction time. - let pool = Db::connect_read_pool(&config, "postgres://user:pw@192.0.2.1:5432/none", 7) + let pool = Db::connect_read_pool(&config, "postgres://user:pw@192.0.2.1:5432/none", 10) .expect("lazy construction must not dial the replica"); - assert_eq!(pool.options().get_max_connections(), 7); - assert_eq!(pool.options().get_min_connections(), 0); + assert_eq!(pool.options().get_max_connections(), 10); + assert_eq!(pool.options().get_min_connections(), 2); + assert!( + !pool.options().get_test_before_acquire(), + "the idle ping must not spend the 150ms reader acquire budget" + ); assert_eq!( pool.options().get_acquire_timeout(), Db::READER_ACQUIRE_TIMEOUT ); + // A nonzero floor must not turn construction into a dial: sqlx 0.9 only + // tops up the floor from the reaper or after a connection closes. + tokio::task::yield_now().await; + assert_eq!(pool.size(), 0, "construction must not dial the replica"); +} + +/// The reader warm floor is a fifth of the ceiling, never zero, and never +/// above the ceiling. +#[test] +fn reader_min_connections_is_a_fifth_of_max_with_a_floor_of_one() { + for (max, min) in [(1, 1), (4, 1), (5, 1), (9, 1), (10, 2), (50, 10), (100, 20)] { + assert_eq!( + Db::reader_min_connections(max), + min, + "max_connections={max}" + ); + } } /// Channel window: head fetch (no cursor) reads the WRITER; cursor pages @@ -1967,7 +1989,7 @@ async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() { }; // `Db::new` so the writer arms the floor guard and the reader is the - // real lazy `connect_read_pool` pool (min_connections=0, 150ms + // real lazy `connect_read_pool` pool (warm floor of 1, 150ms // acquire timeout). Reader is sized 1 so holding one connection // saturates it. let mut db = Db::new(&DbConfig { @@ -2282,7 +2304,7 @@ async fn routed_reads_are_confined_to_the_requested_community() { drop_scratch_db(&admin, writer, &wname).await; } -/// D4: a LAZY reader pool (connect_lazy, min_connections=0, never yet +/// D4: a LAZY reader pool (connect_lazy, never yet /// used) must still let [`Db::spawn_fence_probe`] verify the writer's /// floor guard and spawn — reader-down or reader-idle at boot must not /// disable fence probing. @@ -4097,6 +4119,9 @@ async fn e_tag_any_runs_with_real_binds_on_p_join_and_count() { #[path = "tests/thread_window_postgres_tests.rs"] mod thread_window_postgres_tests; +#[path = "tests/reader_pool_postgres_tests.rs"] +mod reader_pool_postgres_tests; + /// `insert_event_with_serving_write_guard` indexes mentions in the event's own /// transaction, and an indexing failure stays best-effort: the savepoint rolls /// back only the mention rows, never the event. diff --git a/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs new file mode 100644 index 0000000000..a2549715e0 --- /dev/null +++ b/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs @@ -0,0 +1,424 @@ +//! Reader-pool warm floor and fence-proof bounds, against a real Postgres +//! standing in for the replica (the scratch writer database doubles as the +//! reader). + +use super::*; +use std::time::Instant; + +/// A `Db` whose writer and lazy reader both point at one scratch database, +/// with the fence open so routed reads attempt the reader. +async fn same_db_reader(prefix: &str, read_max: u32) -> (PgPool, Db, String, String) { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (seed, wname) = create_scratch_db(&admin, prefix).await; + seed.close().await; + let scratch_url = { + let base = admin_url().await; + let idx = base.rfind('/').expect("db url has a path segment"); + format!("{}/{}", &base[..idx], wname) + }; + let mut db = Db::new(&DbConfig { + database_url: scratch_url.clone(), + read_database_url: Some(scratch_url.clone()), + max_connections: 4, + read_max_connections: Some(read_max), + ..DbConfig::default() + }) + .await + .expect("connect Db with lazy reader"); + db.fence().force_open_for_tests(chrono::Utc::now()); + db.set_replica_read_max_age_for_tests(Some(Duration::from_secs(30))); + (admin, db, wname, scratch_url) +} + +/// Poll until the reader pool holds exactly `target` connections, all idle. +/// Returned and floor-maintenance connections land asynchronously, from +/// tasks sqlx spawns on drop. +async fn await_idle_floor(read_pool: &PgPool, target: u32) { + let deadline = Instant::now() + Duration::from_secs(10); + while read_pool.size() != target || read_pool.num_idle() != target as usize { + assert!( + Instant::now() < deadline, + "reader pool never settled at {target} idle; size={} idle={}", + read_pool.size(), + read_pool.num_idle() + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +/// Counters keyed by `(name, decision, reason)` labels, drained: each +/// snapshot resets the recorder's counters to zero. +fn drain_counters( + snapshotter: &Snapshotter, +) -> std::collections::HashMap<(String, String, String), u64> { + snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter_map(|(key, _, _, value)| { + let DebugValue::Counter(n) = value else { + return None; + }; + let labels: Vec<_> = key.key().labels().collect(); + let get = |name: &str| { + labels + .iter() + .find(|l| l.key() == name) + .map(|l| l.value().to_owned()) + .unwrap_or_default() + }; + Some(( + (key.key().name().to_owned(), get("decision"), get("reason")), + n, + )) + }) + .collect() +} + +/// `buzz_db_route_decision` counters keyed by `(decision, reason)`. +fn route_reasons(snapshotter: &Snapshotter) -> std::collections::HashMap<(String, String), u64> { + drain_counters(snapshotter) + .into_iter() + .filter(|((name, ..), _)| name == "buzz_db_route_decision") + .map(|((_, decision, reason), n)| ((decision, reason), n)) + .collect() +} + +/// `buzz_db_reader_release_closed_total` counters keyed by `reason`. +fn release_closes(snapshotter: &Snapshotter) -> std::collections::HashMap { + drain_counters(snapshotter) + .into_iter() + .filter(|((name, ..), _)| name == "buzz_db_reader_release_closed_total") + .map(|((_, _, reason), n)| (reason, n)) + .collect() +} + +/// The boot ping fills the reader pool to its warm floor, so a freshly +/// started pod meets its first routed reads with established connections +/// instead of dialing each one inside the 150ms reader acquire budget. +/// Without the warm-up the ping leaves exactly one connection behind. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn reader_boot_ping_warms_pool_to_min_connections() { + let (admin, db, wname, _) = same_db_reader("warm_r", 15).await; + let read_pool = db.read_pool.clone().expect("reader pool configured"); + let target = Db::reader_min_connections(15); + assert_eq!( + target, 3, + "fixture needs a floor above the ping's one connection" + ); + assert_eq!( + read_pool.size(), + 0, + "lazy reader dials nothing before the ping" + ); + + db.read_pool_boot_ping_for_tests().await; + + await_idle_floor(&read_pool, target).await; + // Settled at the floor and no further: hold briefly and re-check that + // maintenance does not keep dialing. + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(read_pool.size(), target, "warm-up must stop at the floor"); + + drop_scratch_db(&admin, db.pool.clone(), &wname).await; +} + +/// A warm reader connection whose server stops answering (here: the +/// heartbeat observation blocks behind an exclusive lock) must fail the +/// routed read closed to the writer within the proof budget, not hang the +/// request. Without a pre-acquire ping, the proof is the first round trip +/// on the checkout and carries the liveness bound. +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn routed_read_proof_times_out_to_writer_on_unresponsive_reader() { + let (admin, db, wname, scratch_url) = same_db_reader("proof_to", 5).await; + let read_pool = db.read_pool.clone().expect("reader pool configured"); + db.read_pool_boot_ping_for_tests().await; + await_idle_floor(&read_pool, Db::reader_min_connections(5)).await; + + let mut locker = sqlx::PgConnection::connect(&scratch_url) + .await + .expect("connect locker"); + sqlx::query("BEGIN").execute(&mut locker).await.unwrap(); + sqlx::query("LOCK TABLE replica_heartbeat IN ACCESS EXCLUSIVE MODE") + .execute(&mut locker) + .await + .expect("lock heartbeat"); + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let query = EventQuery::for_community(CommunityId::from_uuid(Uuid::new_v4())); + let start = Instant::now(); + let count = { + let _guard = metrics::set_default_local_recorder(&recorder); + tokio::time::timeout( + Duration::from_secs(10), + db.count_events_routed("proof_timeout_probe", &query), + ) + .await + .expect("routed read must not hang on an unresponsive reader") + } + .expect("writer fallback still answers the read"); + let elapsed = start.elapsed(); + + assert_eq!(count, 0, "writer answered on an empty scratch database"); + assert!( + elapsed >= Db::READER_PROOF_TIMEOUT && elapsed < Db::READER_PROOF_TIMEOUT * 3, + "fallback must wait out one proof budget ({}ms) and no more; took {}ms", + Db::READER_PROOF_TIMEOUT.as_millis(), + elapsed.as_millis() + ); + let reasons = route_reasons(&snapshotter); + assert_eq!( + reasons.get(&("writer".to_owned(), "reader_proof_timeout".to_owned())), + Some(&1), + "unresponsive reader must fall back as writer/reader_proof_timeout; got {reasons:?}" + ); + + sqlx::query("ROLLBACK").execute(&mut locker).await.unwrap(); + locker.close().await.ok(); + drop_scratch_db(&admin, db.pool.clone(), &wname).await; +} + +/// With no pre-acquire ping, a warm reader connection killed server-side is +/// handed out as-is; the routed read must still fail closed to the writer as +/// `reader_validation_error` with no error reaching the caller. +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn terminated_warm_reader_connection_fails_closed_to_writer() { + let (admin, db, wname, _) = same_db_reader("term_r", 5).await; + let read_pool = db.read_pool.clone().expect("reader pool configured"); + db.read_pool_boot_ping_for_tests().await; + await_idle_floor(&read_pool, 1).await; + + let pid: i32 = { + let mut conn = read_pool.acquire().await.expect("warm reader connection"); + sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *conn) + .await + .expect("reader backend pid") + }; + await_idle_floor(&read_pool, 1).await; + let terminated: bool = sqlx::query_scalar("SELECT pg_terminate_backend($1, 5000)") + .bind(pid) + .fetch_one(&admin) + .await + .expect("terminate warm reader backend"); + assert!(terminated, "warm reader backend must be terminated"); + + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let query = EventQuery::for_community(CommunityId::from_uuid(Uuid::new_v4())); + let count = { + let _guard = metrics::set_default_local_recorder(&recorder); + db.count_events_routed("terminated_reader_probe", &query) + .await + } + .expect("a dead reader connection must not surface as a client error"); + + assert_eq!(count, 0, "writer answered on an empty scratch database"); + let reasons = route_reasons(&snapshotter); + assert_eq!( + reasons.get(&("writer".to_owned(), "reader_validation_error".to_owned())), + Some(&1), + "dead warm connection must fall back as writer/reader_validation_error; got {reasons:?}" + ); + + drop_scratch_db(&admin, db.pool.clone(), &wname).await; +} + +/// A proof that times out while its `BEGIN` is still in flight must not +/// return the connection to the pool with a server-side transaction open. +/// sqlx counts the transaction only once `BEGIN` is acknowledged, so a +/// cancelled begin queues no `ROLLBACK`, and the release ping consumes the +/// late `BEGIN` reply: without the reader pool's release check, the +/// connection comes back `idle in transaction` once the reader resumes. +/// The database defaults to `REPEATABLE READ`, the isolation routed reads +/// begin with, so the check only sees the open transaction because reader +/// sessions pin their own default. +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn proof_timeout_mid_begin_leaves_no_reader_transaction_open() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (writer, wname) = create_scratch_db(&admin, "proof_tx").await; + sqlx::query(sqlx::AssertSqlSafe(format!( + "ALTER DATABASE \"{wname}\" SET default_transaction_isolation = 'repeatable read'" + ))) + .execute(&admin) + .await + .expect("default the scratch database to repeatable read"); + let writer_options = writer.connect_options(); + let proxy = crate::test_support::PausingProxy::spawn( + writer_options.get_host().to_owned(), + writer_options.get_port(), + ) + .await; + let read_pool = Db::read_pool_options(5).connect_lazy_with(Db::reader_connect_options( + (*writer_options).clone().host("127.0.0.1").port(proxy.port), + )); + let mut db = Db::from_pools(writer.clone(), read_pool.clone()); + db.fence().force_open_for_tests(chrono::Utc::now()); + db.set_replica_read_max_age_for_tests(Some(Duration::from_secs(30))); + // Primes the Aurora capability cache, so `BEGIN` is the proof's first + // round trip, and warms the floor the routed read will check out. + db.read_pool_boot_ping_for_tests().await; + await_idle_floor(&read_pool, 1).await; + + proxy.set_paused(true); + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let query = EventQuery::for_community(CommunityId::from_uuid(Uuid::new_v4())); + // Held until the release check has run: sqlx runs it in a task spawned + // on this (current-thread) runtime. + let guard = metrics::set_default_local_recorder(&recorder); + let count = db + .count_events_routed("proof_mid_begin_probe", &query) + .await + .expect("writer fallback answers while the reader is paused"); + assert_eq!(count, 0, "writer answered on an empty scratch database"); + let reasons = route_reasons(&snapshotter); + assert_eq!( + reasons.get(&("writer".to_owned(), "reader_proof_timeout".to_owned())), + Some(&1), + "paused reader must fall back as writer/reader_proof_timeout; got {reasons:?}" + ); + + proxy.set_paused(false); + await_idle_floor(&read_pool, 1).await; + let deadline = Instant::now() + Duration::from_secs(2); + loop { + let open: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_stat_activity \ + WHERE datname = $1 AND state = 'idle in transaction'", + ) + .bind(&wname) + .fetch_one(&admin) + .await + .expect("read pg_stat_activity"); + if open == 0 { + break; + } + assert!( + Instant::now() < deadline, + "{open} reader connection(s) returned to the pool idle in transaction" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + drop(guard); + let closes = release_closes(&snapshotter); + assert_eq!( + closes.get("open_transaction"), + Some(&1), + "the release check must close the connection left inside BEGIN; got {closes:?}" + ); + + read_pool.close().await; + drop_scratch_db(&admin, writer, &wname).await; +} + +/// A reader failover whose old sockets stop answering without closing must +/// not strand the reader pool. Each stalled warm connection costs one routed +/// read a proof timeout; once the release check gives up on those sockets, +/// their slots are freed and routed reads reach the reader again through +/// new connections, instead of falling back to the writer until the kernel +/// gives up on the old sockets. +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn stalled_reader_sessions_release_their_slots_for_new_connections() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (writer, wname) = create_scratch_db(&admin, "stall_r").await; + let writer_options = writer.connect_options(); + let proxy = crate::test_support::PausingProxy::spawn( + writer_options.get_host().to_owned(), + writer_options.get_port(), + ) + .await; + // Floor 1, ceiling 2: both slots end up on stalled sessions. + let read_pool = Db::read_pool_options(2).connect_lazy_with(Db::reader_connect_options( + (*writer_options).clone().host("127.0.0.1").port(proxy.port), + )); + let mut db = Db::from_pools(writer.clone(), read_pool.clone()); + db.fence().force_open_for_tests(chrono::Utc::now()); + db.set_replica_read_max_age_for_tests(Some(Duration::from_secs(30))); + db.read_pool_boot_ping_for_tests().await; + await_idle_floor(&read_pool, 1).await; + { + let first = read_pool.acquire().await.expect("first warm connection"); + let second = read_pool.acquire().await.expect("second warm connection"); + drop((first, second)); + } + await_idle_floor(&read_pool, 2).await; + + proxy.stall_established(); + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let query = EventQuery::for_community(CommunityId::from_uuid(Uuid::new_v4())); + let _guard = metrics::set_default_local_recorder(&recorder); + for _ in 0..2 { + db.count_events_routed("stalled_reader_probe", &query) + .await + .expect("writer fallback answers while the reader is stalled"); + } + let reasons = route_reasons(&snapshotter); + assert_eq!( + reasons.get(&("writer".to_owned(), "reader_proof_timeout".to_owned())), + Some(&2), + "each stalled warm connection must cost one proof timeout; got {reasons:?}" + ); + + // The 5s release bound plus slack, as a literal so a loosened bound + // fails here. Snapshots drain the recorder, so totals accumulate. + let stalled_at = Instant::now(); + let bound = Duration::from_secs(8); + let mut totals = std::collections::HashMap::<(String, String, String), u64>::new(); + let served_by_reader = |totals: &std::collections::HashMap<(String, String, String), u64>| { + totals + .iter() + .filter(|((name, decision, _), _)| { + name == "buzz_db_route_decision" && decision == "replica" + }) + .map(|(_, n)| n) + .sum::() + }; + loop { + db.count_events_routed("stalled_reader_probe", &query) + .await + .expect("routed read answers"); + for (key, n) in drain_counters(&snapshotter) { + *totals.entry(key).or_default() += n; + } + if served_by_reader(&totals) > 0 { + break; + } + assert!( + stalled_at.elapsed() < bound, + "routed reads never reached the reader through a new connection within {bound:?}; \ + size={} idle={} counters={totals:?}", + read_pool.size(), + read_pool.num_idle(), + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + let timeout_closes = totals + .get(&( + "buzz_db_reader_release_closed_total".to_owned(), + String::new(), + "timeout".to_owned(), + )) + .copied() + .unwrap_or_default(); + assert!( + timeout_closes >= 1, + "stalled sessions must be closed by the release timeout; got {totals:?}" + ); + drop(_guard); + read_pool.close().await; + drop_scratch_db(&admin, writer, &wname).await; +} diff --git a/crates/buzz-db/src/store/usage.rs b/crates/buzz-db/src/store/usage.rs index 09c0140262..a1532144b3 100644 --- a/crates/buzz-db/src/store/usage.rs +++ b/crates/buzz-db/src/store/usage.rs @@ -1679,74 +1679,6 @@ mod postgres_tests { drop_scratch_db(&admin, writer, &writer_name).await; } - /// Loopback TCP proxy that can make the reader go dark. While dark, any - /// session that receives bytes stops relaying for good but keeps both - /// sockets open, like a replica that stops answering mid-query. Sessions - /// opened after the proxy comes back relay normally. - struct DarkeningProxy { - port: u16, - dark: std::sync::Arc, - } - - impl DarkeningProxy { - async fn spawn(upstream_host: String, upstream_port: u16) -> Self { - use std::sync::atomic::{AtomicBool, Ordering}; - use std::sync::Arc; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - async fn pump( - mut from: tokio::net::tcp::OwnedReadHalf, - mut to: tokio::net::tcp::OwnedWriteHalf, - dark: Arc, - ) { - let mut buf = [0u8; 8192]; - loop { - let Ok(n) = from.read(&mut buf).await else { - return; - }; - if n == 0 { - return; - } - if dark.load(Ordering::SeqCst) { - // Swallow the bytes and hold both sockets open. - let _held = (from, to); - std::future::pending::<()>().await; - return; - } - if to.write_all(&buf[..n]).await.is_err() { - return; - } - } - } - - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") - .await - .expect("bind proxy"); - let port = listener.local_addr().expect("proxy addr").port(); - let dark = Arc::new(AtomicBool::new(false)); - let accept_dark = dark.clone(); - tokio::spawn(async move { - while let Ok((client, _)) = listener.accept().await { - let Ok(server) = - tokio::net::TcpStream::connect((upstream_host.as_str(), upstream_port)) - .await - else { - continue; - }; - let (client_read, client_write) = client.into_split(); - let (server_read, server_write) = server.into_split(); - tokio::spawn(pump(client_read, server_write, accept_dark.clone())); - tokio::spawn(pump(server_read, client_write, accept_dark.clone())); - } - }); - Self { port, dark } - } - - fn set_dark(&self, dark: bool) { - self.dark.store(dark, std::sync::atomic::Ordering::SeqCst); - } - } - /// A fleet collection abandoned at its relay-side deadline on a reader /// that went dark mid-query must release its slot in the shared reader /// pool within SQLx's bounded close-on-drop, not hold it until the kernel @@ -1770,22 +1702,21 @@ mod postgres_tests { insert_user(&reader, reader_community, &random_pubkey(), false).await; let reader_options = reader.connect_options(); - let proxy = DarkeningProxy::spawn( + let proxy = crate::test_support::PausingProxy::spawn( reader_options.get_host().to_owned(), reader_options.get_port(), ) .await; let proxied: PgConnectOptions = (*reader_options).clone().host("127.0.0.1").port(proxy.port); - // One slot, so a stranded slot blocks every later acquire. No - // before-acquire ping, so the stall lands after the checkout is - // handed out, where the relay deadline drops it. - let read_pool = PgPoolOptions::new() - .max_connections(1) + // The production reader options with one slot, so a stranded slot + // blocks every later acquire, and no warm floor, so nothing redials + // it. No before-acquire ping, so the stall lands after the checkout + // is handed out, where the relay deadline drops it. + let read_pool = Db::read_pool_options(1) .min_connections(0) - .test_before_acquire(false) .acquire_timeout(Duration::from_secs(15)) - .connect_with(proxied) + .connect_with(Db::reader_connect_options(proxied)) .await .expect("connect reader through proxy"); let db = Db::from_pools(writer.clone(), read_pool.clone()); @@ -1811,22 +1742,34 @@ mod postgres_tests { // a failed boot ping. This pins close-on-drop ahead of that await. let cold = Db::from_pools(writer.clone(), read_pool.clone()); cold.fence().force_open_for_tests(Utc::now()); - proxy.set_dark(true); - tokio::time::timeout(Duration::from_secs(1), cold.usage_fleet_stock_snapshot()) - .await - .expect_err("a dark reader must stall the collection until its deadline"); - proxy.set_dark(false); + proxy.set_paused(true); + // Under the reader's 1s fence-proof bound, so the caller deadline + // (not the proof timeout) is what abandons the checkout. + tokio::time::timeout( + Duration::from_millis(500), + cold.usage_fleet_stock_snapshot(), + ) + .await + .expect_err("a dark reader must stall the collection until its deadline"); + // Still paused: only close-on-drop can free the slot this fast. The + // reader pool's release check would free it too, but only after its + // 5s timeout, and its follow-up sqlx ping is not bounded at all. let started = Instant::now(); + while read_pool.size() != 0 { + assert!( + started.elapsed() < Duration::from_secs(3), + "slot release must be bounded by close-on-drop, still held after {:?}", + started.elapsed() + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } + proxy.set_paused(false); + let mut replacement = read_pool .acquire() .await - .expect("the abandoned collection must release its reader slot"); - assert!( - started.elapsed() < Duration::from_secs(8), - "slot release must be bounded by close-on-drop, took {:?}", - started.elapsed() - ); + .expect("the released slot must accept a replacement connection"); let one: i32 = sqlx::query_scalar("SELECT 1") .fetch_one(&mut *replacement) .await diff --git a/crates/buzz-db/src/test_support.rs b/crates/buzz-db/src/test_support.rs index 4159e5d83f..62bb89245a 100644 --- a/crates/buzz-db/src/test_support.rs +++ b/crates/buzz-db/src/test_support.rs @@ -314,3 +314,109 @@ pub(crate) async fn wait_for_lock_wait(pool: &sqlx::PgPool, application_name: &s .await .expect("session reached a lock wait"); } + +/// Loopback TCP proxy that can pause the upstream database, like +/// `docker pause` on a replica. While paused, every session (established or +/// new) holds the bytes it receives with both sockets open; on resume the +/// held bytes are delivered in order and the sessions carry on. +/// +/// [`PausingProxy::stall_established`] instead holds only the sessions open +/// at that moment, for good, while later sessions relay normally: a failover +/// whose old sockets stop answering without closing. +pub(crate) struct PausingProxy { + pub(crate) port: u16, + hold: tokio::sync::watch::Sender, + next_session: std::sync::Arc, +} + +/// Which proxy sessions hold their bytes. +#[derive(Clone, Copy, Default)] +struct ProxyHold { + all: bool, + /// Sessions numbered below this hold their bytes. + below: u64, +} + +impl ProxyHold { + fn holds(self, session: u64) -> bool { + self.all || session < self.below + } +} + +impl PausingProxy { + pub(crate) async fn spawn(upstream_host: String, upstream_port: u16) -> Self { + use std::sync::atomic::Ordering; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + async fn pump( + mut from: tokio::net::tcp::OwnedReadHalf, + mut to: tokio::net::tcp::OwnedWriteHalf, + mut hold: tokio::sync::watch::Receiver, + session: u64, + ) { + let mut buf = [0u8; 8192]; + loop { + let Ok(n) = from.read(&mut buf).await else { + return; + }; + if n == 0 { + return; + } + if hold.wait_for(|hold| !hold.holds(session)).await.is_err() { + return; + } + if to.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind proxy"); + let port = listener.local_addr().expect("proxy addr").port(); + let (hold, _) = tokio::sync::watch::channel(ProxyHold::default()); + let accept_hold = hold.subscribe(); + let next_session = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)); + let accept_next = next_session.clone(); + tokio::spawn(async move { + while let Ok((client, _)) = listener.accept().await { + let session = accept_next.fetch_add(1, Ordering::SeqCst); + let Ok(server) = + tokio::net::TcpStream::connect((upstream_host.as_str(), upstream_port)).await + else { + continue; + }; + let (client_read, client_write) = client.into_split(); + let (server_read, server_write) = server.into_split(); + tokio::spawn(pump( + client_read, + server_write, + accept_hold.clone(), + session, + )); + tokio::spawn(pump( + server_read, + client_write, + accept_hold.clone(), + session, + )); + } + }); + Self { + port, + hold, + next_session, + } + } + + pub(crate) fn set_paused(&self, paused: bool) { + self.hold.send_modify(|hold| hold.all = paused); + } + + /// Hold every session open now, for good; later sessions relay normally. + pub(crate) fn stall_established(&self) { + let below = self.next_session.load(std::sync::atomic::Ordering::SeqCst); + self.hold.send_modify(|hold| hold.below = below); + } +} diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index fa181f2a47..9cb1404071 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -538,7 +538,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { info!("Postgres connected (writer + lazy read replica pool)"); // Reader-down at boot must not crash or block the relay; this warn-only // ping is the sole boot-time visibility that the replica is unreachable - // (the lazy pool with min_connections=0 dials nothing until first use). + // (the lazy pool dials nothing at construction). On success it also + // fills the reader's warm floor in the background. db.spawn_read_pool_boot_ping(); } else { info!("Postgres connected"); diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 073bbd0fe0..5f5870e804 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -417,6 +417,21 @@ Pool sizing and the aggregate connection budget across all four roles remain deployment configuration concerns. The relay does not enforce a combined connection ceiling. +The reader pool keeps a warm floor of one fifth of its maximum (at least one) +established connections per relay pod, so standing reader connections are +that floor times the pod count; size the replica's `max_connections` for it. +Each routed read proves the replica is caught up before serving from it. A +proof that does not answer within 1s falls back to the writer, counted as +`buzz_db_route_decision{decision="writer",reason="reader_proof_timeout"}`. +A sustained rise in that series means the replica is reachable but not +answering, for example during a failover. Every connection returned to the +reader pool passes a release check first. A connection is closed instead of +returned, and the pool dials a replacement, when the check does not answer +within 5s (`reason="timeout"`), fails (`error`), or finds a transaction still +open (`open_transaction`). Each close increments +`buzz_db_reader_release_closed_total{reason}`, a counter with those three +series. A rise in `timeout` means reader sockets stopped answering. + ## Relay Pod extensions The chart exposes narrow extension points for init containers, volumes, relay