ecat_data/
rdbms.rs

1// Copyright (c) 2026 erik <erik@erik.xyz> — https://erik.xyz
2use std::time::Duration;
3
4use async_trait::async_trait;
5
6use crate::dialect::Dialect;
7
8#[derive(Debug, Clone)]
9pub struct Row {
10    columns: Vec<String>,
11    values: Vec<serde_json::Value>,
12}
13
14impl Row {
15    /// Create a new Row with the given columns and values.
16    pub fn new(columns: Vec<String>, values: Vec<serde_json::Value>) -> Self {
17        debug_assert_eq!(
18            columns.len(),
19            values.len(),
20            "columns and values must have the same length"
21        );
22        Self { columns, values }
23    }
24
25    pub fn get(&self, col: &str) -> Option<&serde_json::Value> {
26        self.columns
27            .iter()
28            .position(|c| c == col)
29            .and_then(|i| self.values.get(i))
30    }
31
32    pub fn columns(&self) -> &[String] {
33        &self.columns
34    }
35
36    pub fn values(&self) -> &[serde_json::Value] {
37        &self.values
38    }
39
40    /// 按列顺序逐对产出 `(列名, 值)`，用于跨分片合并 / 冷存轮转。
41    pub fn iter(&self) -> impl Iterator<Item = (&str, &serde_json::Value)> {
42        self.columns
43            .iter()
44            .map(String::as_str)
45            .zip(self.values.iter())
46    }
47}
48
49/// 事务内部实现。后端（sqlx / tiberius）实现它，`Transaction` 转发调用。
50#[async_trait]
51pub trait TransactionInner: Send {
52    async fn execute(&mut self, sql: &str) -> Result<u64, RdbmsError>;
53    async fn query(&mut self, sql: &str) -> Result<Vec<Row>, RdbmsError>;
54    async fn execute_with(
55        &mut self,
56        sql: &str,
57        params: &[serde_json::Value],
58    ) -> Result<u64, RdbmsError>;
59    async fn query_with(
60        &mut self,
61        sql: &str,
62        params: &[serde_json::Value],
63    ) -> Result<Vec<Row>, RdbmsError>;
64    fn dialect(&self) -> Dialect;
65    async fn commit(&mut self) -> Result<(), RdbmsError>;
66    async fn rollback(&mut self) -> Result<(), RdbmsError>;
67}
68
69/// 无 backing 连接的事务（[`Transaction::new`]）执行 SQL 时的错误文案。
70/// 这类事务只能作为空占位，执行任何语句都是编程错误 —— 必须报错而非
71/// 静默返回 0 行影响，否则写操作会无声丢失。
72const NO_BACKING: &str = "transaction has no backing connection (created via Transaction::new)";
73
74pub struct Transaction {
75    committed: bool,
76    rolled_back: bool,
77    /// 在 `with_inner` 时从 inner 拷贝，避免 `dialect(&self)` 这个同步方法
78    /// 需要等待异步锁。
79    dialect: Dialect,
80    inner: tokio::sync::Mutex<Option<Box<dyn TransactionInner>>>,
81}
82
83impl Transaction {
84    /// 创建一个无 backing 连接的空事务。
85    ///
86    /// 只能作为占位符用于「不执行任何语句」的场景；在其中执行 SQL 会返回错误。
87    /// 需要真正执行语句时用 [`Transaction::with_inner`] 或后端的 `transaction()`。
88    pub fn new() -> Self {
89        Self {
90            committed: false,
91            rolled_back: false,
92            dialect: Dialect::Standard,
93            inner: tokio::sync::Mutex::new(None),
94        }
95    }
96
97    pub fn with_inner(inner: Box<dyn TransactionInner>) -> Self {
98        let dialect = inner.dialect();
99        Self {
100            committed: false,
101            rolled_back: false,
102            dialect,
103            inner: tokio::sync::Mutex::new(Some(inner)),
104        }
105    }
106
107    pub async fn commit(mut self) -> Result<(), RdbmsError> {
108        if let Some(inner) = self.inner.get_mut().as_mut() {
109            inner.commit().await?;
110        }
111        self.committed = true;
112        Ok(())
113    }
114
115    pub async fn rollback(mut self) -> Result<(), RdbmsError> {
116        if let Some(inner) = self.inner.get_mut().as_mut() {
117            inner.rollback().await?;
118        }
119        self.committed = false;
120        self.rolled_back = true;
121        Ok(())
122    }
123}
124
125impl Default for Transaction {
126    fn default() -> Self {
127        Self::new()
128    }
129}
130
131#[async_trait]
132impl SqlExecutor for Transaction {
133    async fn execute(&self, sql: &str) -> Result<u64, RdbmsError> {
134        let mut guard = self.inner.lock().await;
135        match guard.as_mut() {
136            Some(inner) => inner.execute(sql).await,
137            None => Err(RdbmsError::Database(NO_BACKING.into())),
138        }
139    }
140
141    async fn query(&self, sql: &str) -> Result<Vec<Row>, RdbmsError> {
142        let mut guard = self.inner.lock().await;
143        match guard.as_mut() {
144            Some(inner) => inner.query(sql).await,
145            None => Err(RdbmsError::Database(NO_BACKING.into())),
146        }
147    }
148
149    async fn execute_with(
150        &self,
151        sql: &str,
152        params: &[serde_json::Value],
153    ) -> Result<u64, RdbmsError> {
154        let mut guard = self.inner.lock().await;
155        match guard.as_mut() {
156            Some(inner) => inner.execute_with(sql, params).await,
157            None => Err(RdbmsError::Database(NO_BACKING.into())),
158        }
159    }
160
161    async fn query_with(
162        &self,
163        sql: &str,
164        params: &[serde_json::Value],
165    ) -> Result<Vec<Row>, RdbmsError> {
166        let mut guard = self.inner.lock().await;
167        match guard.as_mut() {
168            Some(inner) => inner.query_with(sql, params).await,
169            None => Err(RdbmsError::Database(NO_BACKING.into())),
170        }
171    }
172
173    /// 在**调用方已有的事务**里跑两条语句：不另开事务，也不提交 ——
174    /// 原子边界归调用方，`insert(&tx, …)` 的语义就是「用我这个事务发这两条」。
175    ///
176    /// 全程持锁：若在两条语句之间放开，并发的 `tx.execute(…)`（同一条连接上的
177    /// 另一个 INSERT）会把 `LAST_INSERT_ID()` 的值换成它插的那条 ——
178    /// 正是本方法要挡的静默错值。
179    async fn execute_then_query(
180        &self,
181        first: &str,
182        first_params: &[serde_json::Value],
183        second: &str,
184    ) -> Result<Vec<Row>, RdbmsError> {
185        let mut guard = self.inner.lock().await;
186        match guard.as_mut() {
187            Some(inner) => {
188                inner.execute_with(first, first_params).await?;
189                inner.query(second).await
190            }
191            None => Err(RdbmsError::Database(NO_BACKING.into())),
192        }
193    }
194
195    fn dialect(&self) -> Dialect {
196        self.dialect
197    }
198}
199
200impl Drop for Transaction {
201    fn drop(&mut self) {
202        // 这里只记日志与计数：Drop 里无法执行异步回滚，实际回滚依赖
203        // 底层 sqlx / tiberius 事务在未提交时 Drop 自动回滚。
204        if !self.committed && !self.rolled_back {
205            crate::timeout::TRANSACTIONS_LEAKED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
206            tracing::warn!("transaction dropped without commit — rolling back");
207        }
208    }
209}
210
211#[async_trait]
212pub trait SqlExecutor: Send + Sync {
213    /// 执行一条 SQL 语句，返回受影响行数。
214    /// 用户提供的值请走 [`SqlExecutor::execute_with`]。
215    async fn execute(&self, sql: &str) -> Result<u64, RdbmsError>;
216    /// 查询多行。用户提供的值请走 [`SqlExecutor::query_with`]。
217    async fn query(&self, sql: &str) -> Result<Vec<Row>, RdbmsError>;
218    /// 参数化执行，防注入。无法绑定参数的后端返回错误。
219    async fn execute_with(
220        &self,
221        _sql: &str,
222        _params: &[serde_json::Value],
223    ) -> Result<u64, RdbmsError> {
224        Err(RdbmsError::Database(
225            "parameterized execute not supported by this backend".into(),
226        ))
227    }
228    /// 参数化查询，防注入。无法绑定参数的后端返回错误。
229    async fn query_with(
230        &self,
231        _sql: &str,
232        _params: &[serde_json::Value],
233    ) -> Result<Vec<Row>, RdbmsError> {
234        Err(RdbmsError::Database(
235            "parameterized query not supported by this backend".into(),
236        ))
237    }
238    /// 写路径且需要返回结果（`INSERT ... RETURNING` / `OUTPUT INSERTED`）。
239    /// 默认委托给 [`SqlExecutor::query_with`]；只有读写分离路由需要覆写，
240    /// 否则写语句会被路由到从库。
241    async fn query_write(
242        &self,
243        sql: &str,
244        params: &[serde_json::Value],
245    ) -> Result<Vec<Row>, RdbmsError> {
246        self.query_with(sql, params).await
247    }
248    /// 在**同一条连接**上先执行 `first`、再执行 `second`（查询），两者原子。
249    ///
250    /// 存在的理由：MySQL 的 `LAST_INSERT_ID()` 是**连接作用域**的 ——
251    /// 池下两次独立取用可能落到不同连接、取回别的会话刚插入的值（静默错值），
252    /// 所以 `insert` 的两步式主键回填必须落在同一条连接上。
253    ///
254    /// 默认实现返回「不支持」。三种覆写：
255    /// - [`Transaction`] —— **直接在自己身上跑两条**（它本来就在一条连接上的事务里）
256    /// - 需要两步式的客户端（如 `ecat-data-sqlx` 的 `SqlxClient`）—— **开一个事务，跑完提交**
257    /// - 一步式后端（MSSQL 的 `OUTPUT INSERTED`、PG/SQLite 的 `RETURNING`）—— 用不到，不覆写
258    async fn execute_then_query(
259        &self,
260        _first: &str,
261        _first_params: &[serde_json::Value],
262        _second: &str,
263    ) -> Result<Vec<Row>, RdbmsError> {
264        Err(RdbmsError::Database(
265            "this backend cannot run two statements atomically on one connection".into(),
266        ))
267    }
268    /// 本执行器背后的数据库方言。
269    fn dialect(&self) -> Dialect;
270}
271
272#[async_trait]
273pub trait RdbmsClient: SqlExecutor {
274    async fn transaction(&self) -> Result<Transaction, RdbmsError>;
275
276    /// 副本延迟探针。`None` = 该后端不提供（默认），或此刻探测不出结果
277    /// （无复制、`Seconds_Behind_*` 为 NULL）；路由按「未知就不跳过」处理。
278    async fn replica_lag(&self) -> Option<Duration> {
279        None
280    }
281
282    /// 该客户端**配置的最大**连接数（预算校验用）；`None` = 后端不报告
283    /// （默认）。预算校验（`ecat-data-sharding` 的 `validate_pool_budget`）
284    /// 对 `None` 按「未知」跳过，不会当成 0 连接。
285    fn pool_size(&self) -> Option<u32> {
286        None
287    }
288}
289
290#[derive(Debug, thiserror::Error)]
291pub enum RdbmsError {
292    #[error("database error: {0}")]
293    Database(String),
294    #[error("connection error: {0}")]
295    Connection(String),
296    #[error("configuration error: {0}")]
297    Config(String),
298    #[error("timeout: {0}")]
299    Timeout(String),
300    /// 读写分离路由：副本全部不可用，且未开启降级读主
301    /// （[`crate::RdbmsRouting`]）。
302    #[error("no available replica")]
303    NoAvailableReplica,
304}
305
306// 测试独立成文件：本文件紧贴 500 行上限（项目硬规则），内联测试会顶过。
307#[cfg(test)]
308mod tests;
ecat_data/rdbms.rs

ecat_data/
rdbms.rs
