From 0e36dbc823f4754b98da64944cc015a9f7acbd6e Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 26 Oct 2026 14:04:39 -0400 Subject: [PATCH] fix(db): keep the reader pool warm and drop its pre-acquire ping (#8495) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary The reader pool's 150ms `acquire_timeout` covers the whole acquire: the pre-acquire idle-connection ping **and** any new-connection dial (TCP + TLS + SCRAM). When readers slow down for a moment, the timeout: - abandons in-flight dials mid-handshake (Postgres logs `could not accept SSL connection: EOF detected`), - closes healthy idle connections whose ping overran instead of returning them to the pool, and - falls every miss back to the writer, which then exhausts its own pool. Across a fleet of relay pods this shows up as a burst of abandoned handshakes on the readers that starts on nearly every pod at once, followed seconds later by writer pool-acquire timeouts. The database itself reports no statement, lock, or connection-limit errors before the burst: the client is tearing down its own reader connections. This PR removes that amplifier. It does not change the request budget, writer fallback, or the writer pool. ### Change (`crates/buzz-db/src/runtime/mod.rs`) - **Warm floor:** reader `min_connections` is now `max / 5`, at least 1, derived from the existing reader pool size (no new config knob). sqlx re-dials the floor in background maintenance, outside the request budget, from its reaper and whenever a connection is closed instead of released. The floor is per relay process, so standing reader connections are the floor times the number of relay processes; size the reader's `max_connections` for it. - **Floor fills at boot:** with `max_lifetime`/`idle_timeout` set, a lazy sqlx 0.9 pool does not dial `min_connections` at construction. The existing boot ping now closes its probe connection instead of returning it, which triggers sqlx's floor maintenance. Probe acquires that overrun 150ms retry with bounded backoff (250ms, 500ms, 1s, 2s). The doc comment that said sqlx dials the minimum eagerly was wrong for this config and is corrected. - **`test_before_acquire(false)` on the reader:** the routed read's `BEGIN … READ ONLY` already proves the connection and fails closed to the writer (`reader_validation_error`). This removes one round trip per checkout and the warm-connection churn. - **Bounded fence proof:** with no pre-acquire ping, the proof (Aurora capability probe, `BEGIN`, heartbeat observation) is the first round trip on a checkout. It now runs under `READER_PROOF_TIMEOUT` (1s), so a reader that stops answering without closing its sockets (partition, host loss, failover) fails closed to the writer as the new route reason `reader_proof_timeout` instead of hanging the request until TCP gives up. 1s is well above a normal round trip, so a brief reader stall waits on the reader rather than flooding the writer. - **Bounded release:** every connection returned to the reader pool first passes a release check bounded at 5s (`Db::read_pool_options`). sqlx's own release ping has no timeout, so without this a socket whose reader went silent would hold its slot until the kernel gives up; with every slot held that way, the pool cannot dial the reader that replaced it, and all routed reads fall back to the writer. A check that times out or errors closes the socket and frees the slot, and floor maintenance dials a replacement. A connection still inside a transaction is closed instead of returned: a checkout dropped while its `BEGIN` was in flight (a timed-out proof, a cancelled caller) has no `ROLLBACK` queued, because sqlx counts the transaction only once `BEGIN` is acknowledged. Closing it ends the session, and the server rolls it back. An expired proof is therefore simply dropped. Maintenance checkouts (fleet usage collection) also keep their close-on-drop: when one is abandoned at its deadline, its socket is closed under sqlx's 5s close bound rather than going through the release path. - **Pinned isolation default:** reader sessions set `default_transaction_isolation=read committed` in the startup packet, which takes precedence over database and role defaults. The release check depends on this, because it detects an open routed-read transaction by its `REPEATABLE READ` isolation. - **Metric:** each close is counted in `buzz_db_reader_release_closed_total{reason}` (`timeout`, `error`, `open_transaction`). Operator notes for the floor, `reader_proof_timeout`, and this counter are in `deploy/charts/buzz/README.md`. ### What got simpler - The reader checkout path does one less round trip and no longer closes healthy connections on a slow ping. - Boot warm-up reuses sqlx's own `min_connections` maintenance rather than re-implementing it. - One release rule now covers every reader checkout, so the open-transaction leak on cancelled reads no longer needs special handling per call site. - No new config knob. - The reader test proxy now lives once in `test_support` instead of inside `store/usage.rs`. - Tests build their reader pools from the production options rather than hand-copied ones. ### Tradeoffs - The reader is now dialed at boot (previously "no reader dial at boot"). If the reader is down at boot, this only produces log noise, and reads still fail closed to the writer. - The release check is one trivial statement (`SELECT current_setting(...)`) per released reader connection. It runs in sqlx's release task, after the response, so request latency is unchanged. - The check detects an open transaction by comparing the current isolation level with the session default. That is only sound because the reader session default is pinned to `READ COMMITTED` and routed reads begin `REPEATABLE READ`. - **Residual window:** after the release check passes, sqlx sends its own release ping (one more round trip), and that ping has no timeout. A serving-read connection whose endpoint goes silent between those two round trips still holds its slot until TCP gives up. Only releases already mid-ping at that instant are exposed: later releases fail the bounded check first and are closed. sqlx behaves the same way for every pool on `main`. Maintenance checkouts avoid it through close-on-drop. Bounding it for serving reads would mean replacing sqlx's release path or putting a pooler in front of the reader, so it is left as a follow-up. - Boot fill is best effort. Each boot-probe retry still uses the 150ms serving acquire budget, so if cold handshakes consistently take longer than that, all five attempts fail and the floor is not filled at boot. It fills the first time any reader connection is closed, which the reaper and the release check both do, and until then reads dial as before. - This also closes pre-existing debt: a caller that dropped a routed read mid-`BEGIN` (client disconnect) could leave a connection `idle in transaction`. The release check covers that path too. ### Related issue N/A (production incident follow-up; none found). ### Testing - Postgres tests (`runtime/tests/reader_pool_postgres_tests.rs`): - the boot ping leaves the reader pool at exactly its floor, all idle (base leaves one connection); - a warm reader whose heartbeat read blocks (exclusive lock) falls back to the writer as `reader_proof_timeout` within one proof budget; - a warm reader connection terminated server-side falls back as `reader_validation_error` with no client error; - a proof that times out mid-`BEGIN` on a paused reader (a test proxy that holds bytes, like `docker pause`) falls back as `reader_proof_timeout`. The reader database default is set to `REPEATABLE READ` for this test. Once the reader resumes, it leaves no `idle in transaction` reader backend, and the close is counted as `open_transaction`. The test fails if the release check is made blind to open transactions, or if the isolation pin is removed. - Failover: every established reader session is stalled while new sessions get through. Each stalled warm connection costs one `reader_proof_timeout`, then routed reads reach the reader again through new connections within 8s, and the closes are counted as `timeout`. The test fails with the release bound removed. - `store/usage.rs`: a fleet collection abandoned at its deadline on a paused reader frees its slot within 3s while the reader is still paused, which is faster than the 5s release check. The test fails with maintenance close-on-drop removed. - Unit tests: the min-connections mapping, the reader pool options, and that construction still dials nothing. - Blox, at `f3e3bf158`: `just fmt-check` and `just clippy` are clean, and `buzz-db` unit tests are 191/191. - Blox, at `01745b21e`: fmt is clean, the targeted reader tests passed 3 of 3 runs, and the release-bound and transaction-blind mutations both turn the targeted tests red. - Blox, at `2f8e482fb` (before the rebase onto `main`): fmt and clippy are clean, the targeted reader tests passed 3 of 3 runs, `buzz-db` unit tests pass, and each of four mutations turns its targeted test red: release bound, transaction-blind check, isolation pin, and maintenance close-on-drop. - Blox, at the rebased head `e80bc2bed`: the same four mutations each turn their targeted test red, and fmt, clippy, the targeted reader tests (3 of 3 runs) and `buzz-db` unit tests are clean. - CI Postgres lane at `e80bc2bed`: 1130/1130 passed, including every new reader-pool test. - Live check at `e80bc2bed`: a relay with a two-slot reader pool behind a proxy, which was then frozen for 18s. All 246 history reads succeeded, and the slowest took 1.0s. During the freeze, 2 proof timeouts and 2 release-timeout closes were counted. Reads returned to the reader within 36ms of the resume, and no reader backend was left `idle in transaction`. - Rebase note: `main` added a reader `after_connect` hook that tightens `statement_timeout`. It now lives in `Db::read_pool_options`, so the tests' reader pools get it too. --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- crates/buzz-db/src/runtime/mod.rs | 327 ++++++++++---- crates/buzz-db/src/runtime/observability.rs | 10 +- crates/buzz-db/src/runtime/tests.rs | 47 +- .../tests/reader_pool_postgres_tests.rs | 424 ++++++++++++++++++ crates/buzz-db/src/store/usage.rs | 117 ++--- crates/buzz-db/src/test_support.rs | 106 +++++ crates/buzz-relay/src/main.rs | 3 +- deploy/charts/buzz/README.md | 15 + 8 files changed, 863 insertions(+), 186 deletions(-) create mode 100644 crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs 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