ecat_data_sqlx/
rdbms.rs

1// Copyright (c) 2026 erik <erik@erik.xyz> — https://erik.xyz
2//! [`RdbmsClient`] 实现：事务开启、副本延迟探测与连接池大小。
3//!
4//! 单独成模块是为了守住「每个源文件 < 500 行」的项目硬规则：`lib.rs`
5//! 已贴着上限，`SqlExecutor` 实现继续留在那里。
6use async_trait::async_trait;
7use ecat_data::{Dialect, RdbmsClient, RdbmsError, SqlExecutor, Transaction, TransactionInner};
8use std::time::Duration;
9
10use crate::pool::Pool;
11use crate::transaction::{MyTx, PgTx, SqTx};
12use crate::{SqlxClient, db_err, mysql_replica_status_sql, parse_mysql_replica_lag};
13
14#[async_trait]
15impl RdbmsClient for SqlxClient {
16    async fn transaction(&self) -> Result<Transaction, RdbmsError> {
17        let dialect = self.pool.dialect();
18        // 超时值随事务一起带走：`transaction()` 之后调用方手里只有 `Transaction`，
19        // 事务 wrapper 是唯一知道该用什么超时的地方。
20        let query_timeout = self.query_timeout;
21        let inner: Box<dyn TransactionInner> = match &self.pool {
22            Pool::Pg(p) => Box::new(PgTx {
23                inner: Some(p.begin().await.map_err(db_err)?),
24                dialect,
25                query_timeout,
26            }),
27            Pool::My(p) => Box::new(MyTx {
28                inner: Some(p.begin().await.map_err(db_err)?),
29                dialect,
30                query_timeout,
31            }),
32            Pool::Sq(p) => Box::new(SqTx {
33                inner: Some(p.begin().await.map_err(db_err)?),
34                dialect,
35                query_timeout,
36            }),
37        };
38        Ok(Transaction::with_inner(inner))
39    }
40
41    /// 只对 MySQL 探测（其余方言返回 `None`）：版本决定用哪条 `SHOW` 语句。
42    async fn replica_lag(&self) -> Option<Duration> {
43        if self.pool.dialect() != Dialect::MySql {
44            return None;
45        }
46        let version = self.mysql_version().await?;
47        let rows = self.query(mysql_replica_status_sql(&version)).await.ok()?;
48        parse_mysql_replica_lag(&rows)
49    }
50
51    /// 配置的最大连接数，不是当前已建立数：预算校验要在建池后立即拿到上限
52    /// （新建池的已建立连接数接近 0，用它校验等于永远通过）。
53    fn pool_size(&self) -> Option<u32> {
54        Some(self.pool.max_connections())
55    }
56}
ecat_data_sqlx/rdbms.rs

ecat_data_sqlx/
rdbms.rs
