fix(deletion): give the scheduled drain the cold-start DB connector (#8146)

## 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-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 }

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.

crates/buzz-db/src/runtime/cold_start.rs

@@ -0,0 +1,204 @@
+//! 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 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)];
+
+/// 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,
+}
+
+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<Self, ColdStartError> {
+        let config = cold_start_config(config);
+        connect_with_retry(worker, || Db::new(&config)).await
+    }
+}
+
+fn cold_start_config(config: DbConfig) -> DbConfig {
+    DbConfig {
+        // 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
+    }
+}
+
+async fn connect_with_retry<T, Connect, Attempt>(
+    worker: &'static str,
+    mut connect: Connect,
+) -> std::result::Result<T, ColdStartError>
+where
+    Connect: FnMut() -> Attempt,
+    Attempt: Future<Output = crate::Result<T>>,
+{
+    let started = Instant::now();
+    // Three attempts of at most 30s, plus 2s and 5s backoffs: at most 97s.
+    for attempt in 0..=RETRY_DELAYS.len() {
+        let attempt_started = Instant::now();
+        eprintln!(
+            "{}",
+            serde_json::json!({
+                "event": format!("{worker}_db_connect_started"),
+                "stage": "db_connect",
+                "attempt": attempt + 1,
+                "timeout_ms": ACQUIRE_TIMEOUT.as_millis(),
+            })
+        );
+        // Bound the entire initialization future, including session setup.
+        let result = timeout(ACQUIRE_TIMEOUT, connect())
+            .await
+            .unwrap_or_else(|_| Err(sqlx::Error::PoolTimedOut.into()));
+        match result {
+            Ok(db) => {
+                eprintln!(
+                    "{}",
+                    serde_json::json!({
+                        "event": format!("{worker}_db_connect_completed"),
+                        "stage": "db_connect",
+                        "attempt": attempt + 1,
+                        "attempt_elapsed_ms": attempt_started.elapsed().as_millis(),
+                        "elapsed_ms": started.elapsed().as_millis(),
+                    })
+                );
+                return Ok(db);
+            }
+            Err(error) => {
+                let delay = RETRY_DELAYS
+                    .get(attempt)
+                    .copied()
+                    .filter(|_| retryable(&error));
+                let class = error_class(&error);
+                // Avoid raw connection errors/URLs: configuration and driver
+                // messages can contain credentials. Keep the source on the
+                // returned error, but print only this bounded classification.
+                eprintln!(
+                    "{}",
+                    serde_json::json!({
+                        "event": format!("{worker}_db_connect_failed"),
+                        "stage": "db_connect",
+                        "attempt": attempt + 1,
+                        "attempt_elapsed_ms": attempt_started.elapsed().as_millis(),
+                        "elapsed_ms": started.elapsed().as_millis(),
+                        "error_class": class,
+                        "io_kind": match &error {
+                            DbError::Sqlx(sqlx::Error::Io(error)) => Some(format!("{:?}", error.kind())),
+                            _ => None,
+                        },
+                        "sqlstate": match &error {
+                            DbError::Sqlx(sqlx::Error::Database(error)) => error.code(),
+                            _ => None,
+                        },
+                        "retry_in_ms": delay.map(|value| value.as_millis()),
+                    })
+                );
+                match delay {
+                    Some(delay) => sleep(delay).await,
+                    None => {
+                        return Err(ColdStartError {
+                            worker,
+                            attempts: attempt + 1,
+                            elapsed_ms: started.elapsed().as_millis(),
+                            class,
+                            source: error,
+                        });
+                    }
+                }
+            }
+        }
+    }
+    unreachable!("the last connection attempt always returns")
+}
+
+fn retryable(error: &DbError) -> bool {
+    match error {
+        DbError::Sqlx(sqlx::Error::PoolTimedOut) => true,
+        // SQLx already backs off on connection refusal and transient server
+        // errors. Other transport/resolver failures may also be transient;
+        // allow only the bounded startup retries, excluding local input and
+        // permission errors. Unknown resolver errors are not labeled as DNS.
+        DbError::Sqlx(sqlx::Error::Io(error)) => !matches!(
+            error.kind(),
+            ErrorKind::InvalidInput
+                | ErrorKind::InvalidData
+                | ErrorKind::PermissionDenied
+                | ErrorKind::NotFound
+                | ErrorKind::Unsupported
+        ),
+        // Includes authentication, TLS, URL configuration and protocol errors
+        // raised while dialing. Session-setup failures arrive as
+        // `PoolTimedOut` (see the module docs).
+        _ => false,
+    }
+}
+
+fn error_class(error: &DbError) -> &'static str {
+    match error {
+        DbError::Sqlx(sqlx::Error::PoolTimedOut) => "timeout",
+        DbError::Sqlx(sqlx::Error::Io(_)) => "io",
+        DbError::Sqlx(sqlx::Error::Tls(_)) => "tls",
+        DbError::Sqlx(sqlx::Error::Configuration(_)) => "configuration",
+        DbError::Sqlx(sqlx::Error::Database(error)) => {
+            if error.code().is_some_and(|code| code.starts_with("28")) {
+                "authentication"
+            } else {
+                "database"
+            }
+        }
+        DbError::Sqlx(sqlx::Error::Protocol(_)) => "protocol",
+        _ => "other",
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/buzz-db/src/runtime/cold_start/tests.rs

@@ -0,0 +1,204 @@
+use std::borrow::Cow;
+use std::cell::Cell;
+
+use super::*;
+
+const WORKER: &str = "test_worker";
+
+#[test]
+fn cold_start_overrides_do_not_change_relay_defaults_pool_size_or_session_policy() {
+    let defaults = DbConfig::default();
+    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, 7);
+    assert_eq!(worker.min_connections, 0);
+    assert_eq!(worker.acquire_timeout_secs, 30);
+    assert_eq!(worker.lock_timeout_ms, 123);
+    assert_eq!(worker.idle_txn_timeout_ms, 456);
+    assert_eq!(worker.statement_timeout_ms, 789);
+    assert_eq!(defaults.max_connections, 20);
+    assert_eq!(defaults.min_connections, 2);
+    assert_eq!(defaults.acquire_timeout_secs, 3);
+}
+
+#[tokio::test(start_paused = true)]
+async fn cold_connection_can_exceed_the_old_three_second_budget() {
+    let started = Instant::now();
+    connect_with_retry(WORKER, || async {
+        sleep(Duration::from_secs(6)).await;
+        Ok(())
+    })
+    .await
+    .expect("cold startup succeeds");
+    assert_eq!(started.elapsed(), Duration::from_secs(6));
+}
+
+#[tokio::test(start_paused = true)]
+async fn transient_failure_backs_off_before_a_successful_attempt() {
+    let attempts = Cell::new(0);
+    let started = Instant::now();
+    let result = connect_with_retry(WORKER, || {
+        attempts.set(attempts.get() + 1);
+        let attempt = attempts.get();
+        async move {
+            if attempt == 1 {
+                Err(sqlx::Error::Io(ErrorKind::ConnectionReset.into()).into())
+            } else {
+                Ok(42)
+            }
+        }
+    })
+    .await;
+    assert_eq!(result.expect("second attempt succeeds"), 42);
+    assert_eq!(attempts.get(), 2);
+    assert_eq!(started.elapsed(), Duration::from_secs(2));
+}
+
+#[tokio::test(start_paused = true)]
+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(WORKER, || {
+        attempts.set(attempts.get() + 1);
+        async { Err::<(), _>(sqlx::Error::Io(ErrorKind::ConnectionReset.into()).into()) }
+    })
+    .await
+    .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.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(WORKER, || {
+        attempts.set(attempts.get() + 1);
+        std::future::pending::<crate::Result<()>>()
+    })
+    .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.source,
+        DbError::Sqlx(sqlx::Error::PoolTimedOut)
+    ));
+}
+
+#[tokio::test(start_paused = true)]
+async fn permanent_errors_fail_immediately_and_do_not_expose_connection_details() {
+    let attempts = Cell::new(0);
+    let started = Instant::now();
+    for driver_error in [
+        sqlx::Error::Configuration("sensitive-url".into()),
+        sqlx::Error::Tls("sensitive-certificate".into()),
+        sqlx::Error::Protocol("sensitive-server-reply".into()),
+        sqlx::Error::Io(ErrorKind::PermissionDenied.into()),
+    ] {
+        let mut error = Some(driver_error);
+        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()) }
+        })
+        .await
+        .expect_err("permanent error");
+        assert!(!result.to_string().contains("sensitive"));
+    }
+    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<Cow<'_, str>> {
+        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<Self>) -> Box<dyn std::error::Error + Send + Sync + 'static> {
+        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:?}"
+        );
+    }
+}

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,