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

ecat_data_sqlx/
config.rs
