systemprompt_database/services/postgres/
connection.rs1use std::future::Future;
14use std::time::Duration;
15
16use sqlx::postgres::{PgConnectOptions, PgPool, PgPoolOptions};
17
18use crate::error::DatabaseResult;
19use crate::resilience::classify::Outcome;
20use crate::resilience::config::RetryConfig;
21use crate::resilience::retry::retry_async;
22
23const RETRY_DELAYS_MS: &[u64] = &[100, 200, 400, 800, 1600];
24const MAX_ATTEMPTS: u32 = 5;
25
26#[derive(Debug, Clone, Copy)]
33pub struct PoolConfig {
34 pub max_connections: u32,
35 pub min_connections: u32,
36 pub acquire_timeout: Duration,
37 pub idle_timeout: Duration,
38 pub max_lifetime: Duration,
39}
40
41impl Default for PoolConfig {
42 fn default() -> Self {
43 Self {
44 max_connections: 50,
45 min_connections: 0,
46 acquire_timeout: Duration::from_secs(30),
47 idle_timeout: Duration::from_mins(5),
48 max_lifetime: Duration::from_mins(30),
49 }
50 }
51}
52
53#[must_use]
54pub fn build_pool_options(cfg: &PoolConfig) -> PgPoolOptions {
55 PgPoolOptions::new()
56 .max_connections(cfg.max_connections)
57 .min_connections(cfg.min_connections)
58 .max_lifetime(cfg.max_lifetime)
59 .acquire_timeout(cfg.acquire_timeout)
60 .idle_timeout(cfg.idle_timeout)
61}
62
63pub async fn connect_with_retry(
64 options: PgPoolOptions,
65 connect_options: PgConnectOptions,
66) -> DatabaseResult<PgPool> {
67 let connector = |opts: PgConnectOptions| {
68 let options = options.clone();
69 async move { options.connect_with(opts).await }
70 };
71 connect_with_retry_using(connect_options, MAX_ATTEMPTS, RETRY_DELAYS_MS, connector).await
72}
73
74pub async fn connect_with_retry_using<T, F, Fut>(
75 connect_options: PgConnectOptions,
76 max_attempts: u32,
77 delays_ms: &[u64],
78 connector: F,
79) -> DatabaseResult<T>
80where
81 T: Send,
82 F: Fn(PgConnectOptions) -> Fut + Send + Sync,
83 Fut: Future<Output = Result<T, sqlx::Error>> + Send,
84{
85 let cfg = RetryConfig {
86 max_attempts,
87 base_delay: Duration::from_millis(delays_ms.first().copied().unwrap_or(100)),
88 max_delay: Duration::from_millis(delays_ms.iter().copied().max().unwrap_or(1600)),
89 jitter: false,
90 };
91 let classify = |err: &sqlx::Error| {
92 if is_retryable(err) {
93 Outcome::Transient { retry_after: None }
94 } else {
95 Outcome::Permanent
96 }
97 };
98 retry_async(&cfg, "postgres-connect", classify, || {
99 connector(connect_options.clone())
100 })
101 .await
102 .map_err(Into::into)
103}
104
105fn is_retryable(err: &sqlx::Error) -> bool {
106 if let sqlx::Error::Io(io_err) = err
107 && io_err.kind() == std::io::ErrorKind::ConnectionRefused
108 {
109 return true;
110 }
111 let msg = err.to_string();
112 msg.contains("unexpected response from SSLRequest") || msg.contains("starting up")
113}