ecat_data_sqlx/
config.rs

1// Copyright (c) 2026 erik <erik@erik.xyz> — https://erik.xyz
2
3use ecat_data::Dialect;
4use ecat_tls::TlsClientConfig;
5use serde::Deserialize;
6use std::time::Duration;
7
8#[derive(Debug, Clone, Deserialize)]
9pub struct SqlxConfig {
10    pub url: String,
11    #[serde(default)]
12    pub username: Option<String>,
13    #[serde(default)]
14    pub password: Option<String>,
15    /// **当前不支持**。SQLx 的 TLS 走 URL 参数（如 `?sslmode=require`）；
16    /// 设了本字段会在 [`crate::SqlxClient::from_config`] 处**报错**，
17    /// 而不是静默忽略。
18    #[serde(default)]
19    pub tls: Option<TlsClientConfig>,
20    /// 会话初始化语句；未配置时按方言给默认值，见
21    /// [`SqlxConfig::effective_session_init`]。
22    #[serde(default)]
23    pub session_init: Option<Vec<String>>,
24
25    #[serde(default)]
26    pub max_connections: Option<u32>,
27    #[serde(default)]
28    pub min_connections: Option<u32>,
29    /// 取连接的最长等待。注意 **`0` 不是「禁用」** —— sqlx 对
30    /// `acquire_timeout(Duration::ZERO)` 的处理是**立即超时**（拿不到连接就失败），
31    /// 与 [`SqlxConfig::query_timeout_secs`] 的 `0`（禁用）语义不同。
32    #[serde(default)]
33    pub acquire_timeout_secs: Option<u64>,
34    #[serde(default)]
35    pub idle_timeout_secs: Option<u64>,
36    #[serde(default)]
37    pub max_lifetime_secs: Option<u64>,
38    /// `0` = 禁用查询超时。
39    #[serde(default)]
40    pub query_timeout_secs: Option<u64>,
41    /// 每次从池取连接是否先 ping。默认 `false`：sqlx 默认在**每次** acquire
42    /// 都 ping（不论上次 ping 多近，见 sqlx issue #1743），在高频路径上是
43    /// 一笔无谓往返。关掉后由 `max_lifetime` + 查询超时 + 首次使用报错兜底。
44    #[serde(default)]
45    pub test_before_acquire: Option<bool>,
46}
47
48/// 池参数的解析结果（带默认值）。由 [`SqlxConfig::pool`] 产出。
49#[derive(Debug, Clone)]
50pub struct PoolParams {
51    pub max_connections: u32,
52    pub min_connections: u32,
53    pub acquire_timeout: Duration,
54    pub idle_timeout: Duration,
55    pub max_lifetime: Duration,
56    pub query_timeout: Option<Duration>,
57    pub test_before_acquire: bool,
58    /// 每条新连接建立后执行的会话初始化语句。
59    pub session_init: Vec<String>,
60}
61
62impl Default for PoolParams {
63    /// 结构体默认值：池大小/超时按文档默认，`session_init` 为**空**
64    /// （不带任何会话设置）。要「按方言的默认」请用 [`PoolParams::for_url`]。
65    fn default() -> Self {
66        Self {
67            max_connections: 10,
68            min_connections: 0,
69            acquire_timeout: Duration::from_secs(30),
70            idle_timeout: Duration::from_secs(600),
71            max_lifetime: Duration::from_secs(1800),
72            query_timeout: Some(Duration::from_secs(30)),
73            test_before_acquire: false,
74            session_init: Vec::new(),
75        }
76    }
77}
78
79impl PoolParams {
80    /// 按 URL 的方言给出「未配置时」的默认参数 —— **含方言默认的 `session_init`**。
81    ///
82    /// 与 [`PoolParams::default`] 的区别：`Default` 是结构体默认值（无会话初始化），
83    /// 本方法才是「用户没配任何东西时应当得到什么」。
84    /// `connect()` 与 `from_config()` 都必须走这一条，否则两个构造器行为会静默不一致。
85    pub fn for_url(url: &str) -> Self {
86        Self {
87            session_init: dialect_session_init(url),
88            ..Self::default()
89        }
90    }
91}
92
93/// 未显式配置时按方言给默认会话初始化语句（库侧时区设为 UTC，
94/// 与 ORM 的「时间统一 UTC」约定对齐）。SQLite 无会话概念，返回空。
95pub fn dialect_session_init(url: &str) -> Vec<String> {
96    match Dialect::from_url(url) {
97        Dialect::Postgres => vec![
98            "SET TIME ZONE 'UTC'".to_string(),
99            "SET application_name = 'ecat'".to_string(),
100        ],
101        Dialect::MySql => vec!["SET time_zone = '+00:00'".to_string()],
102        _ => Vec::new(),
103    }
104}
105
106impl SqlxConfig {
107    /// 未配置的字段一律取 [`PoolParams::default`] —— 默认值只此一处定义，
108    /// 避免与 [`PoolParams::for_url`] 静默分叉。
109    pub fn pool(&self) -> PoolParams {
110        let d = PoolParams::default();
111        let max_connections = self.max_connections.unwrap_or(d.max_connections);
112        PoolParams {
113            max_connections,
114            // min 必须 ≤ max：sqlx 自己的后台保底是「尽力而为」—— 拿不到 permit 就
115            // 安静收手（`sqlx-core-0.8.6/src/pool/inner.rs:396` 的 `try_min_connections`），
116            // 但 `warm_up()` 是**持着已取的连接**再去 acquire 到 min 条，池上限更低
117            // 时永远取不满，只会阻塞到 acquire_timeout 后报错。
118            min_connections: self
119                .min_connections
120                .unwrap_or(d.min_connections)
121                .min(max_connections),
122            acquire_timeout: self
123                .acquire_timeout_secs
124                .map_or(d.acquire_timeout, Duration::from_secs),
125            idle_timeout: self
126                .idle_timeout_secs
127                .map_or(d.idle_timeout, Duration::from_secs),
128            max_lifetime: self
129                .max_lifetime_secs
130                .map_or(d.max_lifetime, Duration::from_secs),
131            query_timeout: self.query_timeout(),
132            test_before_acquire: self.test_before_acquire.unwrap_or(d.test_before_acquire),
133            session_init: self.effective_session_init(),
134        }
135    }
136
137    /// `0` 表示显式禁用超时；未配置时为 30 秒。
138    pub fn query_timeout(&self) -> Option<Duration> {
139        match self.query_timeout_secs {
140            None => Some(Duration::from_secs(30)),
141            Some(0) => None,
142            Some(s) => Some(Duration::from_secs(s)),
143        }
144    }
145
146    /// 会话初始化语句：显式配置优先，未配置时按方言取默认
147    /// （见 `dialect_session_init`）。
148    ///
149    /// 显式空数组会覆盖方言默认，即主动关闭会话初始化
150    /// （与 `query_timeout_secs: 0` 表示禁用同族）。
151    pub fn effective_session_init(&self) -> Vec<String> {
152        self.session_init
153            .clone()
154            .unwrap_or_else(|| dialect_session_init(&self.url))
155    }
156}
157
158#[cfg(test)]
159mod tests {
160    use super::*;
161
162    #[test]
163    fn minimal_config_uses_documented_defaults() {
164        let cfg: SqlxConfig =
165            serde_json::from_str(r#"{"url": "postgres://localhost/db"}"#).unwrap();
166        let p = cfg.pool();
167        assert_eq!(p.max_connections, 10);
168        assert_eq!(p.min_connections, 0);
169        assert_eq!(p.acquire_timeout, Duration::from_secs(30));
170        assert_eq!(p.idle_timeout, Duration::from_secs(600));
171        assert_eq!(p.max_lifetime, Duration::from_secs(1800));
172        assert_eq!(p.query_timeout, Some(Duration::from_secs(30)));
173        assert!(!p.test_before_acquire);
174        assert!(cfg.session_init.is_none());
175    }
176
177    #[test]
178    fn pool_params_can_be_overridden() {
179        let cfg: SqlxConfig = serde_json::from_str(
180            r#"{"url": "mysql://localhost/db",
181                "max_connections": 32,
182                "query_timeout_secs": 0,
183                "test_before_acquire": true,
184                "session_init": ["SET time_zone = '+00:00'"]}"#,
185        )
186        .unwrap();
187        assert_eq!(cfg.pool().max_connections, 32);
188        assert_eq!(cfg.pool().query_timeout, None);
189        assert!(cfg.pool().test_before_acquire);
190        assert_eq!(
191            cfg.effective_session_init(),
192            vec!["SET time_zone = '+00:00'".to_string()]
193        );
194    }
195
196    /// 未显式配置时按方言给默认会话设置：库侧直接返回 UTC，
197    /// 与 ORM 的「时间统一 UTC」约定对齐。
198    #[test]
199    fn session_init_defaults_per_dialect() {
200        let pg: SqlxConfig = serde_json::from_str(r#"{"url": "postgres://h/db"}"#).unwrap();
201        assert!(
202            pg.effective_session_init()
203                .iter()
204                .any(|s| s.contains("TIME ZONE"))
205        );
206
207        let sqlite: SqlxConfig = serde_json::from_str(r#"{"url": "sqlite::memory:"}"#).unwrap();
208        assert!(sqlite.effective_session_init().is_empty());
209    }
210
211    /// 未配置时 `for_url`（connect 走它）与 `pool()`（from_config 走它）
212    /// 必须逐字段一致 —— 否则两个构造器静默分叉，且不会有任何测试变红。
213    #[test]
214    fn for_url_matches_pool_defaults_field_by_field() {
215        for url in ["postgres://h/db", "mysql://h/db", "sqlite::memory:"] {
216            let cfg: SqlxConfig = serde_json::from_str(&format!(r#"{{"url": "{url}"}}"#)).unwrap();
217            let a = PoolParams::for_url(url);
218            let b = cfg.pool();
219            assert_eq!(a.max_connections, b.max_connections, "{url}");
220            assert_eq!(a.min_connections, b.min_connections, "{url}");
221            assert_eq!(a.acquire_timeout, b.acquire_timeout, "{url}");
222            assert_eq!(a.idle_timeout, b.idle_timeout, "{url}");
223            assert_eq!(a.max_lifetime, b.max_lifetime, "{url}");
224            assert_eq!(a.query_timeout, b.query_timeout, "{url}");
225            assert_eq!(a.test_before_acquire, b.test_before_acquire, "{url}");
226            assert_eq!(a.session_init, b.session_init, "{url}");
227        }
228    }
229
230    /// `min > max` 必须夹到 max：否则 `warm_up()` 阻塞到 acquire_timeout 才报错。
231    #[test]
232    fn min_connections_is_clamped_to_max() {
233        let cfg: SqlxConfig = serde_json::from_str(
234            r#"{"url": "sqlite::memory:", "max_connections": 2, "min_connections": 5}"#,
235        )
236        .unwrap();
237        let p = cfg.pool();
238        assert_eq!(p.max_connections, 2);
239        assert_eq!(p.min_connections, 2);
240    }
241
242    /// `query_timeout_secs: 0` 是「禁用」而非「0 秒立刻超时」。
243    #[test]
244    fn zero_query_timeout_means_disabled() {
245        let cfg: SqlxConfig =
246            serde_json::from_str(r#"{"url": "sqlite::memory:", "query_timeout_secs": 0}"#).unwrap();
247        assert_eq!(cfg.query_timeout(), None);
248    }
249
250    /// 设了 `tls` 必须响亮失败 —— 曾静默忽略，与「不伪造/不静默」的原则冲突。
251    #[tokio::test]
252    async fn tls_field_is_rejected_not_silently_ignored() {
253        let cfg: SqlxConfig =
254            serde_json::from_str(r#"{"url": "postgres://h/db", "tls": {"skip_verify": true}}"#)
255                .unwrap();
256        assert!(
257            cfg.tls.is_some(),
258            "serde 应当照旧接受该字段（报错发生在 from_config）"
259        );
260
261        // 不用 unwrap_err（SqlxClient 未实现 Debug，为测试给它加 derive 是本末倒置）
262        let Err(err) = crate::SqlxClient::from_config(cfg).await else {
263            panic!("tls 字段设了必须报错，不能静默忽略");
264        };
265        let msg = err.to_string();
266        assert!(msg.contains("tls"), "got: {msg}");
267        assert!(
268            msg.contains("sslmode"),
269            "错误信息要指向正确做法，got: {msg}"
270        );
271    }
272}
ecat_data_sqlx/config.rs

ecat_data_sqlx/
config.rs
