From c6340a45a072c1793280cdcaec793d4a5dd82d97 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 8 Oct 2026 10:30:38 -0400 Subject: [PATCH] fix(deletion): give the scheduled drain the cold-start DB connector (#8146) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary The scheduled `buzz-admin deletions drain` Job never gets past its first database connection in a hosted staging deployment. Every run exits 5 after exactly 3 seconds with `pool timed out while waiting for an open connection`. The drain starts in a fresh pod with no mesh sidecar, and that pod's first connection can take longer than the relay's 3-second acquisition budget. `connect_store()` in `buzz-deletion` inherits that budget from `DbConfig::default()`, with no retry. In the same pod, the same command run 15 seconds later succeeds. TCP to Postgres, Redis, and S3 from a sidecar-less pod is fine. #7770 fixed the same failure for `buzz-admin storage-snapshot`, but only inside `buzz-admin`. This PR moves that connector into `buzz-db` as `Db::connect_cold_start` and uses it from both run-once workers: - **Pool settings:** a 30s acquisition budget, and no idle spare connections (`min_connections: 0`). - **Retries:** startup timeouts and transient transport errors retry at most twice, after 2s and 5s, so startup waits at most 97s. - **No retry:** configuration, authentication, TLS, and protocol errors fail immediately, and the error message stays free of credentials. - **Per-worker settings:** each worker keeps its own pool size. Storage-snapshot stays at 1 connection, and deletion stays at `BUZZ_DB_POOL_SIZE` (default 20). Each worker also keeps its session timeouts. - **Startup events:** the stderr events carry a worker prefix. Storage-snapshot still emits `storage_snapshot_db_connect_*`, and the deletion worker now emits `deletion_db_connect_*`. - **Error text:** the exhaustion message is now `{worker} database startup failed after …`. For storage-snapshot that reads `storage_snapshot database startup failed …` instead of `storage snapshot database startup failed …`. Nothing in this repo matches on that string. - **Session-setup failures:** SQLx retries a failed `after_connect` hook (writer floor guard, session timeouts, isolation assertion) until the acquire deadline. Those failures therefore surface as `timeout`, are retried, and fail closed after at most 97s instead of immediately. Relay pool defaults are unchanged. The deletion CLI's operator commands (`list`, `approve`, …) use the same connector. They normally run inside the warm relay container, where the first attempt succeeds. **What got simpler:** there is now one cold-start connector instead of a `buzz-admin`-private copy that the deletion worker couldn't reach. `storage_snapshot_startup.rs` is gone, and its unit tests moved with the code. The two Postgres startup tests now share one harness with the drain's tests (`postgres_storage_snapshot.rs` → `postgres_cold_start.rs`). ### Related issue Follow-up to #7770 (same failure, other worker) and #7827 (the drain CronJob). No duplicate found. ### Testing All at head `1545de7b5`: - **GitHub CI:** green on attempt 1, including `PostgreSQL Domain / PostgreSQL Tests`. - **Blox:** - fmt, `clippy --workspace --all-targets -D warnings`, and the `buzz-db` / `buzz-admin` / `buzz-deletion` tests pass. - All four `postgres_cold_start` tests pass, each against a freshly migrated database. - Falsification: with main's `connect_store` restored, the drain retry test fails. - **Agent review:** two rounds. All findings are fixed or disclosed; the error-text change is listed above. - **Agent exercise:** real `buzz-admin deletions list` / `deletions drain` against local Postgres 17 behind a fault proxy. - **5s delay:** succeeds on attempt 1 (3/3). Main fails at about 3.5s. - **Dropped handshake, then 5s delay:** attempt 1 fails with `UnexpectedEof`, then a 2s backoff, then attempt 2 succeeds (3/3). - **Wrong password, missing database or malformed URL:** exit 5 after one attempt. - **Persistent rejection:** exit 5 after three attempts, with 2s and 5s backoffs. - The deletion queue stayed empty throughout. - **Not covered:** an approved deletion, real S3 / Pod Identity, TLS and session-setup failures, and a live 97s stall. --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-admin/src/main.rs | 27 +++- ...age_snapshot.rs => postgres_cold_start.rs} | 89 ++++++++++++- crates/buzz-db/Cargo.toml | 2 +- crates/buzz-db/src/lib.rs | 5 +- .../src/runtime/cold_start.rs} | 99 +++++++++++---- .../src/runtime/cold_start}/tests.rs | 117 +++++++++++++++--- crates/buzz-db/src/runtime/mod.rs | 2 + crates/buzz-deletion/src/lib.rs | 7 +- 8 files changed, 297 insertions(+), 51 deletions(-) rename crates/buzz-admin/tests/{postgres_storage_snapshot.rs => postgres_cold_start.rs} (78%) rename crates/{buzz-admin/src/storage_snapshot_startup.rs => buzz-db/src/runtime/cold_start.rs} (58%) rename crates/{buzz-admin/src/storage_snapshot_startup => buzz-db/src/runtime/cold_start}/tests.rs (50%) diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 8700f926df..312f01c1de 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -22,7 +22,6 @@ mod communities; mod deletions; -mod storage_snapshot_startup; use std::future::Future; use std::sync::atomic::{AtomicU64, Ordering}; @@ -196,6 +195,15 @@ async fn run(cli: Cli) -> Result { } } +/// One session: the worker detaches its lock-owning connection, and the +/// cold-start pool opens no idle replacements while it scans S3. +fn storage_snapshot_db_config(base: DbConfig) -> DbConfig { + DbConfig { + max_connections: 1, + ..base + } +} + async fn cmd_storage_snapshot(max_objects: u64) -> Result { let max_objects_db = i64::try_from(max_objects) .map_err(|_| anyhow::anyhow!("--max-objects must be at most {}", i64::MAX))?; @@ -203,7 +211,11 @@ async fn cmd_storage_snapshot(max_objects: u64) -> Result { return Err(anyhow::anyhow!("--max-objects must be greater than zero")); } - let db = storage_snapshot_startup::connect_db().await?; + let db = Db::connect_cold_start( + storage_snapshot_db_config(db_config_from_env()), + "storage_snapshot", + ) + .await?; let mut leader = db.try_lock_storage_accounting().await?.ok_or_else(|| { anyhow::anyhow!("another storage-snapshot worker already holds the lease") })?; @@ -887,6 +899,17 @@ mod storage_snapshot_tests { use super::*; + #[test] + fn storage_snapshot_holds_a_single_database_session() { + let config = storage_snapshot_db_config(DbConfig { + max_connections: 20, + lock_timeout_ms: 123, + ..DbConfig::default() + }); + assert_eq!(config.max_connections, 1); + assert_eq!(config.lock_timeout_ms, 123); + } + #[tokio::test] async fn failed_fold_never_invokes_snapshot_persistence() { let persist_calls = Arc::new(AtomicUsize::new(0)); diff --git a/crates/buzz-admin/tests/postgres_storage_snapshot.rs b/crates/buzz-admin/tests/postgres_cold_start.rs similarity index 78% rename from crates/buzz-admin/tests/postgres_storage_snapshot.rs rename to crates/buzz-admin/tests/postgres_cold_start.rs index 79b960983e..5307fe532b 100644 --- a/crates/buzz-admin/tests/postgres_storage_snapshot.rs +++ b/crates/buzz-admin/tests/postgres_cold_start.rs @@ -1,4 +1,4 @@ -//! Run the real command through delayed/broken PostgreSQL connections. +//! Run the real run-once workers through delayed/broken PostgreSQL connections. //! The PostgreSQL nextest wrapper supplies a separate database per test. use std::process::{ExitStatus, Stdio}; @@ -164,6 +164,39 @@ struct WorkerOutput { } async fn run_worker(database_url: &str, proxy: &Server, s3: &Server) -> WorkerOutput { + run_admin( + database_url, + proxy, + &["storage-snapshot", "--max-objects", "100"], + &[ + ("BUZZ_S3_ENDPOINT", format!("http://{}", s3.address)), + ("BUZZ_STORAGE_SNAPSHOT_CODE_SHA", "startup-test".to_owned()), + ], + ) + .await +} + +/// Run the scheduled deletion drain against an idle queue. Its Redis pool is +/// lazy and an idle drain never reaches S3, so neither endpoint needs to exist. +async fn run_drain(database_url: &str, proxy: &Server) -> WorkerOutput { + run_admin( + database_url, + proxy, + &["deletions", "drain", "--executor-id", "cold-start-test"], + &[ + ("BUZZ_S3_ENDPOINT", "http://127.0.0.1:9".to_owned()), + ("REDIS_URL", "redis://127.0.0.1:9".to_owned()), + ], + ) + .await +} + +async fn run_admin( + database_url: &str, + proxy: &Server, + args: &[&str], + env: &[(&str, String)], +) -> WorkerOutput { let mut url = Url::parse(database_url).expect("database URL"); url.set_host(Some("127.0.0.1")).expect("proxy host"); url.set_port(Some(proxy.address.port())) @@ -176,15 +209,14 @@ async fn run_worker(database_url: &str, proxy: &Server, s3: &Server) -> WorkerOu let worker_binary = std::env::var_os("NEXTEST_BIN_EXE_buzz_admin") .unwrap_or_else(|| env!("CARGO_BIN_EXE_buzz-admin").into()); let mut child = Command::new(worker_binary) - .args(["storage-snapshot", "--max-objects", "100"]) + .args(args) .env_clear() .env("DATABASE_URL", url.as_str()) - .env("BUZZ_S3_ENDPOINT", format!("http://{}", s3.address)) .env("BUZZ_S3_BUCKET", "test-bucket") .env("BUZZ_S3_REGION", "us-east-1") .env("BUZZ_S3_ACCESS_KEY", "test-access-key") .env("BUZZ_S3_SECRET_KEY", "test-secret-key") - .env("BUZZ_STORAGE_SNAPSHOT_CODE_SHA", "startup-test") + .envs(env.iter().map(|(key, value)| (key, value))) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true) @@ -290,3 +322,52 @@ async fn storage_snapshot_exhaustion_never_lists_s3_or_replaces_the_last_good_sn assert_eq!(snapshot, json!({"previous": "complete"})); assert_eq!(revision, "before"); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn deletion_drain_retries_then_connects_after_the_old_three_second_budget() { + let (url, _pool) = database().await; + let proxy = postgres_proxy(&url, 1, Duration::from_secs(5)).await; + let output = run_drain(&url, &proxy).await; + assert!( + output.status.success(), + "{}\n{}", + output.stdout, + output.stderr + ); + // The rejected dial plus the successful retry. The drain's first claim + // can race SQLx's asynchronous release of the startup connection and dial + // one more session, so the count is a lower bound; the startup events + // below are what pin the retry. + assert!(proxy.requests.load(Ordering::SeqCst) >= 2); + let startup = events(&output.stderr); + assert!(startup + .iter() + .any(|event| event["event"] == "deletion_db_connect_failed" + && event["attempt"] == 1 + && event["retry_in_ms"] == 2000)); + assert!(startup + .iter() + .any(|event| event["event"] == "deletion_db_connect_completed" + && event["attempt"] == 2 + && event["attempt_elapsed_ms"].as_u64().unwrap() >= 5000)); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn deletion_drain_exhaustion_fails_after_three_bounded_attempts() { + let (url, _pool) = database().await; + let proxy = postgres_proxy(&url, usize::MAX, Duration::ZERO).await; + let output = run_drain(&url, &proxy).await; + assert_eq!(output.status.code(), Some(5), "{}", output.stderr); + assert_eq!(proxy.requests.load(Ordering::SeqCst), 3); + let failures: Vec<_> = events(&output.stderr) + .into_iter() + .filter(|event| event["event"] == "deletion_db_connect_failed") + .collect(); + assert_eq!(failures.len(), 3); + assert!(failures.last().unwrap()["retry_in_ms"].is_null()); + assert!(output + .stderr + .contains("deletion database startup failed after 3 attempt(s)")); +} diff --git a/crates/buzz-db/Cargo.toml b/crates/buzz-db/Cargo.toml index 47c8691bf0..7e080808ac 100644 --- a/crates/buzz-db/Cargo.toml +++ b/crates/buzz-db/Cargo.toml @@ -26,5 +26,5 @@ rand = { workspace = true } metrics = { workspace = true } [dev-dependencies] -tokio = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } metrics-util = { workspace = true } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 0a3a1ea398..f8adfefab7 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -31,8 +31,9 @@ pub mod error; mod test_support; pub use runtime::{ - insert_mentions, migration, replica_fence, AdmittedTx, Db, DbConfig, DbConnectionOutcome, - DbConnectionStep, DbPoolRole, DbPoolStats, DbReadinessOutcome, ReadSession, + insert_mentions, migration, replica_fence, AdmittedTx, ColdStartError, Db, DbConfig, + DbConnectionOutcome, DbConnectionStep, DbPoolRole, DbPoolStats, DbReadinessOutcome, + ReadSession, }; /// Valid low-cardinality `(pool_role, operation)` pairs for pool-acquisition telemetry. diff --git a/crates/buzz-admin/src/storage_snapshot_startup.rs b/crates/buzz-db/src/runtime/cold_start.rs similarity index 58% rename from crates/buzz-admin/src/storage_snapshot_startup.rs rename to crates/buzz-db/src/runtime/cold_start.rs index 1a9955ed1b..5ddcd3ccf1 100644 --- a/crates/buzz-admin/src/storage_snapshot_startup.rs +++ b/crates/buzz-db/src/runtime/cold_start.rs @@ -1,39 +1,86 @@ -//! Bounded cold database startup for the run-once storage accounting worker. +//! Bounded cold database startup for run-once operator workers. +//! +//! A freshly scheduled worker pod (no mesh sidecar, cold DNS) can need more +//! than the relay's three-second acquisition budget to open its first +//! connection. Workers that start, do bounded work, and exit share this +//! connector so a slow first connection is waited out and transient transport +//! failures are retried, while configuration, authentication, TLS, and +//! protocol errors during the dial still fail immediately. SQLx treats a +//! failed session-setup hook (`after_connect`) as retryable until the +//! acquire deadline, so such failures surface here as timeouts and are +//! retried within the same bounded budget. +//! +//! Startup attempts are reported as JSON lines on stderr because these +//! command-line workers do not install a tracing subscriber. use std::future::Future; use std::io::ErrorKind; use std::time::Duration; -use anyhow::{Context, Result}; -use buzz_db::{Db, DbConfig, DbError}; use tokio::time::{sleep, timeout, Instant}; +use super::{Db, DbConfig}; +use crate::DbError; + const ACQUIRE_TIMEOUT: Duration = Duration::from_secs(30); const RETRY_DELAYS: [Duration; 2] = [Duration::from_secs(2), Duration::from_secs(5)]; -/// Establish the worker's database pool before it acquires its accounting lease. -pub(super) async fn connect_db() -> Result { - let config = worker_config(crate::db_config_from_env()); - connect_with_retry(|| Db::new(&config)).await +/// A run-once worker could not establish its database pool. +/// +/// The message carries only a bounded classification: driver messages and +/// URLs can contain credentials. The driver error stays available as the +/// [`std::error::Error::source`]. +#[derive(Debug, thiserror::Error)] +#[error( + "{worker} database startup failed after {attempts} attempt(s) and {elapsed_ms}ms ({class})" +)] +pub struct ColdStartError { + worker: &'static str, + attempts: usize, + elapsed_ms: u128, + class: &'static str, + #[source] + source: DbError, } -fn worker_config(config: DbConfig) -> DbConfig { +impl Db { + /// Connect a run-once worker's pool with a bounded cold-start budget. + /// + /// The pool keeps the caller's size and session policy, opens no idle + /// spare connections, and waits up to 30 seconds per acquisition. Startup + /// timeouts and transient transport failures are retried at most twice, + /// after two and five seconds, so startup waits at most 97 seconds. + /// `worker` prefixes the stderr startup events, for example + /// `deletion_db_connect_failed`. + /// + /// Only initial connection establishment is retried. Once a caller holds + /// a session-bound lock or lease, reconnecting would lose that fence. + pub async fn connect_cold_start( + config: DbConfig, + worker: &'static str, + ) -> std::result::Result { + let config = cold_start_config(config); + connect_with_retry(worker, || Db::new(&config)).await + } +} + +fn cold_start_config(config: DbConfig) -> DbConfig { DbConfig { - max_connections: 1, - // The worker detaches its lock-owning session. Do not open idle - // replacements while that session scans S3 and publishes the result. + // A run-once worker only needs connections it actually uses; idle + // replacements would also contend with a slow cold start. min_connections: 0, acquire_timeout_secs: ACQUIRE_TIMEOUT.as_secs(), ..config } } -// Only initial connection establishment is retried. Once the command owns -// the advisory lock, reconnecting would lose its publication fence. -async fn connect_with_retry(mut connect: Connect) -> Result +async fn connect_with_retry( + worker: &'static str, + mut connect: Connect, +) -> std::result::Result where Connect: FnMut() -> Attempt, - Attempt: Future>, + Attempt: Future>, { let started = Instant::now(); // Three attempts of at most 30s, plus 2s and 5s backoffs: at most 97s. @@ -42,7 +89,7 @@ where eprintln!( "{}", serde_json::json!({ - "event": "storage_snapshot_db_connect_started", + "event": format!("{worker}_db_connect_started"), "stage": "db_connect", "attempt": attempt + 1, "timeout_ms": ACQUIRE_TIMEOUT.as_millis(), @@ -57,7 +104,7 @@ where eprintln!( "{}", serde_json::json!({ - "event": "storage_snapshot_db_connect_completed", + "event": format!("{worker}_db_connect_completed"), "stage": "db_connect", "attempt": attempt + 1, "attempt_elapsed_ms": attempt_started.elapsed().as_millis(), @@ -78,7 +125,7 @@ where eprintln!( "{}", serde_json::json!({ - "event": "storage_snapshot_db_connect_failed", + "event": format!("{worker}_db_connect_failed"), "stage": "db_connect", "attempt": attempt + 1, "attempt_elapsed_ms": attempt_started.elapsed().as_millis(), @@ -98,12 +145,12 @@ where match delay { Some(delay) => sleep(delay).await, None => { - return Err(error).with_context(|| { - format!( - "storage snapshot database startup failed after {} attempt(s) and {}ms ({class})", - attempt + 1, - started.elapsed().as_millis(), - ) + return Err(ColdStartError { + worker, + attempts: attempt + 1, + elapsed_ms: started.elapsed().as_millis(), + class, + source: error, }); } } @@ -128,7 +175,9 @@ fn retryable(error: &DbError) -> bool { | ErrorKind::NotFound | ErrorKind::Unsupported ), - // Includes authentication, TLS, URL configuration and protocol errors. + // Includes authentication, TLS, URL configuration and protocol errors + // raised while dialing. Session-setup failures arrive as + // `PoolTimedOut` (see the module docs). _ => false, } } diff --git a/crates/buzz-admin/src/storage_snapshot_startup/tests.rs b/crates/buzz-db/src/runtime/cold_start/tests.rs similarity index 50% rename from crates/buzz-admin/src/storage_snapshot_startup/tests.rs rename to crates/buzz-db/src/runtime/cold_start/tests.rs index e69f2c8a61..5060c7c980 100644 --- a/crates/buzz-admin/src/storage_snapshot_startup/tests.rs +++ b/crates/buzz-db/src/runtime/cold_start/tests.rs @@ -1,17 +1,21 @@ +use std::borrow::Cow; use std::cell::Cell; use super::*; +const WORKER: &str = "test_worker"; + #[test] -fn worker_pool_overrides_do_not_change_relay_defaults_or_session_policy() { +fn cold_start_overrides_do_not_change_relay_defaults_pool_size_or_session_policy() { let defaults = DbConfig::default(); - let worker = worker_config(DbConfig { + let worker = cold_start_config(DbConfig { + max_connections: 7, lock_timeout_ms: 123, idle_txn_timeout_ms: 456, statement_timeout_ms: 789, ..defaults.clone() }); - assert_eq!(worker.max_connections, 1); + assert_eq!(worker.max_connections, 7); assert_eq!(worker.min_connections, 0); assert_eq!(worker.acquire_timeout_secs, 30); assert_eq!(worker.lock_timeout_ms, 123); @@ -25,7 +29,7 @@ fn worker_pool_overrides_do_not_change_relay_defaults_or_session_policy() { #[tokio::test(start_paused = true)] async fn cold_connection_can_exceed_the_old_three_second_budget() { let started = Instant::now(); - connect_with_retry(|| async { + connect_with_retry(WORKER, || async { sleep(Duration::from_secs(6)).await; Ok(()) }) @@ -38,7 +42,7 @@ async fn cold_connection_can_exceed_the_old_three_second_budget() { async fn transient_failure_backs_off_before_a_successful_attempt() { let attempts = Cell::new(0); let started = Instant::now(); - let result = connect_with_retry(|| { + let result = connect_with_retry(WORKER, || { attempts.set(attempts.get() + 1); let attempt = attempts.get(); async move { @@ -59,7 +63,7 @@ async fn transient_failure_backs_off_before_a_successful_attempt() { async fn retry_exhaustion_propagates_the_last_error_after_three_attempts() { let attempts = Cell::new(0); let started = Instant::now(); - let error = connect_with_retry(|| { + let error = connect_with_retry(WORKER, || { attempts.set(attempts.get() + 1); async { Err::<(), _>(sqlx::Error::Io(ErrorKind::ConnectionReset.into()).into()) } }) @@ -67,28 +71,28 @@ async fn retry_exhaustion_propagates_the_last_error_after_three_attempts() { .expect_err("exhausted startup must fail"); assert_eq!(attempts.get(), 3); assert_eq!(started.elapsed(), Duration::from_secs(7)); + assert!(error + .to_string() + .starts_with("test_worker database startup")); assert!(error.to_string().contains("3 attempt(s)")); - assert!(matches!( - error.downcast_ref::(), - Some(DbError::Sqlx(sqlx::Error::Io(_))) - )); + assert!(matches!(error.source, DbError::Sqlx(sqlx::Error::Io(_)))); } #[tokio::test(start_paused = true)] async fn hung_initialization_is_bounded_to_97_seconds_including_backoff() { let attempts = Cell::new(0); let started = Instant::now(); - let error = connect_with_retry(|| { + let error = connect_with_retry(WORKER, || { attempts.set(attempts.get() + 1); - std::future::pending::>() + std::future::pending::>() }) .await .expect_err("hung connection setup must time out"); assert_eq!(attempts.get(), 3); assert_eq!(started.elapsed(), Duration::from_secs(97)); assert!(matches!( - error.downcast_ref::(), - Some(DbError::Sqlx(sqlx::Error::PoolTimedOut)) + error.source, + DbError::Sqlx(sqlx::Error::PoolTimedOut) )); } @@ -103,7 +107,7 @@ async fn permanent_errors_fail_immediately_and_do_not_expose_connection_details( sqlx::Error::Io(ErrorKind::PermissionDenied.into()), ] { let mut error = Some(driver_error); - let result = connect_with_retry(|| { + let result = connect_with_retry(WORKER, || { attempts.set(attempts.get() + 1); let error = error.take().expect("permanent errors must not retry"); async move { Err::<(), _>(error.into()) } @@ -115,3 +119,86 @@ async fn permanent_errors_fail_immediately_and_do_not_expose_connection_details( assert_eq!(attempts.get(), 4); assert_eq!(started.elapsed(), Duration::ZERO); } + +#[derive(Debug, thiserror::Error)] +#[error("sensitive server message")] +struct ServerError(&'static str); + +impl sqlx::error::DatabaseError for ServerError { + fn message(&self) -> &str { + "sensitive server message" + } + + fn code(&self) -> Option> { + Some(Cow::Borrowed(self.0)) + } + + fn as_error(&self) -> &(dyn std::error::Error + Send + Sync + 'static) { + self + } + + fn as_error_mut(&mut self) -> &mut (dyn std::error::Error + Send + Sync + 'static) { + self + } + + fn into_error(self: Box) -> Box { + self + } + + fn kind(&self) -> sqlx::error::ErrorKind { + sqlx::error::ErrorKind::Other + } +} + +#[test] +fn classifier_table_covers_every_retry_and_class_arm() { + let io = |kind: ErrorKind| DbError::Sqlx(sqlx::Error::Io(kind.into())); + let cases: Vec<(DbError, bool, &str)> = vec![ + (DbError::Sqlx(sqlx::Error::PoolTimedOut), true, "timeout"), + (io(ErrorKind::ConnectionReset), true, "io"), + (io(ErrorKind::ConnectionRefused), true, "io"), + (io(ErrorKind::ConnectionAborted), true, "io"), + (io(ErrorKind::TimedOut), true, "io"), + (io(ErrorKind::UnexpectedEof), true, "io"), + (io(ErrorKind::Other), true, "io"), + (io(ErrorKind::InvalidInput), false, "io"), + (io(ErrorKind::InvalidData), false, "io"), + (io(ErrorKind::PermissionDenied), false, "io"), + (io(ErrorKind::NotFound), false, "io"), + (io(ErrorKind::Unsupported), false, "io"), + (DbError::Sqlx(sqlx::Error::Tls("x".into())), false, "tls"), + ( + DbError::Sqlx(sqlx::Error::Configuration("x".into())), + false, + "configuration", + ), + ( + DbError::Sqlx(sqlx::Error::Protocol("x".into())), + false, + "protocol", + ), + ( + DbError::Sqlx(sqlx::Error::Database(Box::new(ServerError("28P01")))), + false, + "authentication", + ), + ( + DbError::Sqlx(sqlx::Error::Database(Box::new(ServerError("28000")))), + false, + "authentication", + ), + ( + DbError::Sqlx(sqlx::Error::Database(Box::new(ServerError("53300")))), + false, + "database", + ), + (DbError::Sqlx(sqlx::Error::RowNotFound), false, "other"), + ]; + for (error, retry, class) in cases { + assert_eq!( + (retryable(&error), error_class(&error)), + (retry, class), + "{error:?}" + ); + } +} diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 5932945f0d..3369919f1e 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -1,10 +1,12 @@ mod admitted_tx; +mod cold_start; mod connection_observability; pub mod migration; pub(crate) mod observability; pub mod replica_fence; pub use admitted_tx::AdmittedTx; +pub use cold_start::ColdStartError; pub use connection_observability::{DbConnectionOutcome, DbConnectionStep}; pub(crate) use connection_observability::{ CONNECTION_DURATION_STEPS, CONNECTION_RAW_SERIES_PER_POD, CONNECTION_STARTED_STEPS, diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index 9e96f64715..4e226755ce 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -547,13 +547,16 @@ fn resolve_submit_host(host: Option<&str>, relay_url: Option<&str>) -> Result Result { let database_url = required_env("DATABASE_URL")?; - let db = Db::new( - &DbConfig { + // The scheduled drain starts in a fresh pod whose first connection can + // outlast the relay's three-second budget. + let db = Db::connect_cold_start( + DbConfig { database_url, max_connections: env_parse("BUZZ_DB_POOL_SIZE", 20), ..DbConfig::default() } .with_session_timeouts_from_env(), + "deletion", ) .await?; Ok(store(&db))