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

ecat_data_sqlx/
config.rs
