sz_orm_core/
transaction.rs

1//! Transaction support
2//!
3//! Provides ACID transaction management
4
5use crate::error::{TransactionState, TxError};
6use crate::pool::Connection;
7use std::sync::Arc;
8use std::time::{Duration, Instant};
9use tokio::sync::Mutex;
10
11// TransactionState 定义在 `error` 模块以避免 `transaction` ↔ `error` 循环依赖；
12// 通过 `pub use error::*;` 在 crate 根重导出，外部访问路径仍为 `sz_orm_core::TransactionState`。
13
14/// 事务隔离级别
15#[derive(Debug, Clone, PartialEq, Default)]
16pub enum IsolationLevel {
17    /// 读未提交
18    ReadUncommitted,
19    /// 读已提交
20    ReadCommitted,
21    /// 可重复读（默认）
22    #[default]
23    RepeatableRead,
24    /// 串行化
25    Serializable,
26    /// 快照隔离
27    Snapshot,
28}
29
30impl std::fmt::Display for IsolationLevel {
31    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32        match self {
33            IsolationLevel::ReadUncommitted => write!(f, "READ UNCOMMITTED"),
34            IsolationLevel::ReadCommitted => write!(f, "READ COMMITTED"),
35            IsolationLevel::RepeatableRead => write!(f, "REPEATABLE READ"),
36            IsolationLevel::Serializable => write!(f, "SERIALIZABLE"),
37            IsolationLevel::Snapshot => write!(f, "SNAPSHOT"),
38        }
39    }
40}
41
42/// 自动提交模式
43#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
44pub enum AutoCommit {
45    /// 开启自动提交（默认）
46    #[default]
47    On,
48    /// 关闭自动提交
49    Off,
50}
51
52/// 事务传播行为
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
54pub enum PropagationBehavior {
55    /// 如果当前存在事务，则加入；否则新建事务（默认）
56    #[default]
57    Required,
58    /// 必须在事务中执行，否则抛出异常
59    Mandatory,
60    /// 必须在没有事务的环境中执行，否则抛出异常
61    Never,
62    /// 如果当前存在事务，则加入；否则无事务执行
63    Supports,
64    /// 总是新建事务，挂起当前事务
65    RequiresNew,
66    /// 如果当前存在事务，则嵌套执行（保存点）；否则新建事务
67    Nested,
68}
69
70/// 事务执行选项
71pub struct TransactOptions {
72    /// 隔离级别
73    pub isolation_level: Option<IsolationLevel>,
74    /// 是否只读
75    pub read_only: bool,
76    /// 超时时间
77    pub timeout: Option<Duration>,
78    /// H-8 修复：嵌套事务（保存点）最大深度限制
79    ///
80    /// 默认 `DEFAULT_MAX_NESTING_DEPTH`（8），防止递归事务导致数据库连接耗尽或
81    /// 保存点栈溢出。设为 0 表示禁用嵌套事务（首次 `savepoint()` 即报错）。
82    pub max_nesting_depth: u32,
83    /// 事务传播行为（默认 Required）
84    pub propagation: PropagationBehavior,
85}
86
87/// H-8 默认最大嵌套深度
88pub const DEFAULT_MAX_NESTING_DEPTH: u32 = 8;
89
90impl Default for TransactOptions {
91    fn default() -> Self {
92        Self {
93            isolation_level: None,
94            read_only: false,
95            timeout: None,
96            max_nesting_depth: DEFAULT_MAX_NESTING_DEPTH,
97            propagation: PropagationBehavior::default(),
98        }
99    }
100}
101
102impl TransactOptions {
103    /// 设置隔离级别
104    pub fn with_isolation(mut self, level: IsolationLevel) -> Self {
105        self.isolation_level = Some(level);
106        self
107    }
108
109    /// 设置为只读事务
110    pub fn read_only(mut self) -> Self {
111        self.read_only = true;
112        self
113    }
114
115    /// 设置事务超时时间
116    pub fn with_timeout(mut self, timeout: Duration) -> Self {
117        self.timeout = Some(timeout);
118        self
119    }
120
121    /// H-8 修复：设置最大嵌套深度
122    pub fn with_max_nesting_depth(mut self, max_depth: u32) -> Self {
123        self.max_nesting_depth = max_depth;
124        self
125    }
126
127    /// 设置事务传播行为
128    pub fn with_propagation(mut self, propagation: PropagationBehavior) -> Self {
129        self.propagation = propagation;
130        self
131    }
132}
133
134/// 校验保存点名称（防止 SQL 注入）
135///
136/// 保存点名称规则：
137/// - 非空
138/// - 只能包含 ASCII 字母、数字、下划线
139/// - 不能以数字开头
140fn validate_savepoint_name(name: &str) -> Result<(), TxError> {
141    if name.is_empty() {
142        return Err(TxError::InvalidSavepointName(name.to_string()));
143    }
144    if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
145        return Err(TxError::InvalidSavepointName(name.to_string()));
146    }
147    if name.starts_with(|c: char| c.is_ascii_digit()) {
148        return Err(TxError::InvalidSavepointName(name.to_string()));
149    }
150    Ok(())
151}
152
153/// 事务对象，封装一个数据库事务
154///
155/// 内部连接以 `Option<Box<dyn Connection>>` 形式持有：
156/// - 事务执行期间，连接存在
157/// - 调用 `take_connection()` 可在 commit/rollback 后取回连接归还到连接池
158/// - Drop 时若事务仍 Active，会尝试 spawn 后台 rollback 任务
159pub struct Transaction {
160    conn: Arc<Mutex<Option<Box<dyn Connection>>>>,
161    state: TransactionState,
162    options: TransactOptions,
163    savepoint_counter: u32,
164    /// 事务超时截止时间（由 options.timeout 计算，在 commit/check 时检查）
165    deadline: Option<Instant>,
166}
167
168impl Transaction {
169    /// 创建新事务（调用方应先通过 connection.begin_transaction() 启动事务）
170    ///
171    /// L-5 修复：补充示例文档
172    ///
173    /// 通常通过 `Connection::begin_transaction()` 创建，而不是直接调用此方法。
174    ///
175    /// # 示例
176    ///
177    /// ```ignore
178    /// use sz_orm_core::transaction::{Transaction, TransactOptions};
179    ///
180    /// # async fn example(conn: Box<dyn sz_orm_core::pool::Connection>) {
181    /// // 通常通过 Connection::begin_transaction 创建
182    /// let tx = Transaction::new(conn, TransactOptions::default());
183    /// assert!(tx.is_active());
184    /// # }
185    /// ```
186    pub fn new(conn: Box<dyn Connection>, options: TransactOptions) -> Self {
187        // 任务1：根据 options.timeout 计算事务截止时间
188        let deadline = options.timeout.map(|t| Instant::now() + t);
189        Self {
190            conn: Arc::new(Mutex::new(Some(conn))),
191            state: TransactionState::Active,
192            options,
193            savepoint_counter: 0,
194            deadline,
195        }
196    }
197
198    /// 获取当前事务状态
199    pub fn state(&self) -> TransactionState {
200        self.state
201    }
202
203    /// 检查事务是否仍然活跃
204    pub fn is_active(&self) -> bool {
205        self.state == TransactionState::Active
206    }
207
208    /// 提交事务
209    ///
210    /// L-5 修复：补充示例文档
211    ///
212    /// 提交后事务状态变为 `Committed`，不可再次 commit/rollback。
213    /// 若未提交就 drop，会自动 rollback。
214    ///
215    /// # 示例
216    ///
217    /// ```ignore
218    /// # use sz_orm_core::transaction::TransactOptions;
219    /// # async fn example(mut tx: sz_orm_core::transaction::Transaction) -> Result<(), Box<dyn std::error::Error>> {
220    /// tx.execute("INSERT INTO users (name) VALUES ('Alice')").await?;
221    /// tx.execute("INSERT INTO users (name) VALUES ('Bob')").await?;
222    /// tx.commit().await?; // 提交两条 INSERT
223    /// # Ok(())
224    /// # }
225    /// ```
226    pub async fn commit(&mut self) -> Result<(), TxError> {
227        if self.state != TransactionState::Active {
228            return Err(TxError::NotActive(self.state));
229        }
230        // 任务1：检查事务超时，超时则回滚并返回错误
231        if let Some(deadline) = self.deadline {
232            if Instant::now() > deadline {
233                self.rollback().await.ok();
234                return Err(TxError::CommitFailed("Transaction timeout".to_string()));
235            }
236        }
237        let mut conn_guard = self.conn.lock().await;
238        let conn = conn_guard.as_mut().ok_or(TxError::ConnectionTaken)?;
239        conn.commit()
240            .await
241            .map_err(|e| TxError::CommitFailed(e.to_string()))?;
242        self.state = TransactionState::Committed;
243        Ok(())
244    }
245
246    /// 回滚事务
247    pub async fn rollback(&mut self) -> Result<(), TxError> {
248        if self.state != TransactionState::Active {
249            return Err(TxError::NotActive(self.state));
250        }
251        let mut conn_guard = self.conn.lock().await;
252        let conn = conn_guard.as_mut().ok_or(TxError::ConnectionTaken)?;
253        conn.rollback()
254            .await
255            .map_err(|e| TxError::RollbackFailed(e.to_string()))?;
256        self.state = TransactionState::RolledBack;
257        Ok(())
258    }
259
260    /// 在事务中执行 SQL（在事务未结束时执行）
261    pub async fn execute(&mut self, sql: &str) -> Result<u64, TxError> {
262        if self.state != TransactionState::Active {
263            return Err(TxError::NotActive(self.state));
264        }
265        let mut conn_guard = self.conn.lock().await;
266        let conn = conn_guard.as_mut().ok_or(TxError::ConnectionTaken)?;
267        let result = conn
268            .execute(sql)
269            .await
270            .map_err(|e| TxError::CommitFailed(e.to_string()))?;
271        Ok(result)
272    }
273
274    /// 在事务中执行查询
275    pub async fn query(
276        &mut self,
277        sql: &str,
278    ) -> Result<Vec<std::collections::HashMap<String, crate::value::Value>>, TxError> {
279        if self.state != TransactionState::Active {
280            return Err(TxError::NotActive(self.state));
281        }
282        let mut conn_guard = self.conn.lock().await;
283        let conn = conn_guard.as_mut().ok_or(TxError::ConnectionTaken)?;
284        let result = conn
285            .query(sql)
286            .await
287            .map_err(|e| TxError::CommitFailed(e.to_string()))?;
288        Ok(result)
289    }
290
291    /// 创建保存点（用于嵌套事务）
292    ///
293    /// 返回自动生成的保存点名（格式 `sp_<N>`，N 单调递增）。
294    ///
295    /// H-8 修复：检查嵌套深度，超过 `options.max_nesting_depth` 时返回
296    /// `TxError::MaxNestingDepthExceeded`。
297    ///
298    /// L-5 修复：补充示例文档
299    ///
300    /// # 示例
301    ///
302    /// ```ignore
303    /// # async fn example(mut tx: sz_orm_core::transaction::Transaction) -> Result<(), Box<dyn std::error::Error>> {
304    /// // 在外层事务中创建保存点
305    /// let sp = tx.savepoint().await?;
306    /// // 执行一些操作
307    /// tx.execute("INSERT INTO orders (id) VALUES (1)").await?;
308    /// // 出错时回滚到保存点（不影响外层事务的其他操作）
309    /// tx.rollback_to_savepoint(&sp).await?;
310    /// // 不再需要保存点时释放
311    /// tx.release_savepoint(&sp).await?;
312    /// # Ok(())
313    /// # }
314    /// ```
315    pub async fn savepoint(&mut self) -> Result<String, TxError> {
316        if self.state != TransactionState::Active {
317            return Err(TxError::NotActive(self.state));
318        }
319        // H-8 修复：嵌套深度检查
320        // savepoint_counter 表示已创建的保存点数；新保存点的深度为 counter + 1
321        let next_depth = self.savepoint_counter + 1;
322        if next_depth > self.options.max_nesting_depth {
323            return Err(TxError::MaxNestingDepthExceeded {
324                current_depth: next_depth,
325                max_depth: self.options.max_nesting_depth,
326            });
327        }
328        self.savepoint_counter += 1;
329        let name = format!("sp_{}", self.savepoint_counter);
330        // 内部生成的名称已通过命名规则（sp_ + 数字），但为防御性编程仍校验
331        validate_savepoint_name(&name)?;
332        let sql = format!("SAVEPOINT {}", name);
333        let mut conn_guard = self.conn.lock().await;
334        let conn = conn_guard.as_mut().ok_or(TxError::ConnectionTaken)?;
335        conn.execute(&sql)
336            .await
337            .map_err(|e| TxError::SavepointError(e.to_string()))?;
338        Ok(name)
339    }
340
341    /// 回滚到保存点
342    ///
343    /// `name` 必须是合法的保存点名称（仅 ASCII 字母/数字/下划线，且不以数字开头）。
344    /// 通常使用 `savepoint()` 返回的名称。
345    pub async fn rollback_to_savepoint(&mut self, name: &str) -> Result<(), TxError> {
346        if self.state != TransactionState::Active {
347            return Err(TxError::NotActive(self.state));
348        }
349        validate_savepoint_name(name)?;
350        let sql = format!("ROLLBACK TO SAVEPOINT {}", name);
351        let mut conn_guard = self.conn.lock().await;
352        let conn = conn_guard.as_mut().ok_or(TxError::ConnectionTaken)?;
353        conn.execute(&sql)
354            .await
355            .map_err(|e| TxError::SavepointError(e.to_string()))?;
356        Ok(())
357    }
358
359    /// 释放保存点
360    ///
361    /// `name` 必须是合法的保存点名称（仅 ASCII 字母/数字/下划线，且不以数字开头）。
362    /// 通常使用 `savepoint()` 返回的名称。
363    pub async fn release_savepoint(&mut self, name: &str) -> Result<(), TxError> {
364        if self.state != TransactionState::Active {
365            return Err(TxError::NotActive(self.state));
366        }
367        validate_savepoint_name(name)?;
368        let sql = format!("RELEASE SAVEPOINT {}", name);
369        let mut conn_guard = self.conn.lock().await;
370        let conn = conn_guard.as_mut().ok_or(TxError::ConnectionTaken)?;
371        conn.execute(&sql)
372            .await
373            .map_err(|e| TxError::SavepointError(e.to_string()))?;
374        Ok(())
375    }
376
377    /// 取出底层连接（用于归还到连接池）
378    ///
379    /// 仅在事务已 commit/rollback 后才能调用，否则返回 `NotActive` 错误。
380    /// 重复调用返回 `ConnectionTaken` 错误。
381    ///
382    /// 典型用法：
383    /// ```ignore
384    /// tx.commit().await?;
385    /// let conn = tx.take_connection().await?;
386    /// pool.release(conn).await;
387    /// ```
388    pub async fn take_connection(&mut self) -> Result<Box<dyn Connection>, TxError> {
389        if self.state == TransactionState::Active {
390            return Err(TxError::NotActive(self.state));
391        }
392        let mut conn_guard = self.conn.lock().await;
393        conn_guard.take().ok_or(TxError::ConnectionTaken)
394    }
395
396    /// 获取事务选项
397    pub fn options(&self) -> &TransactOptions {
398        &self.options
399    }
400}
401
402/// M-8 修复：检测错误字符串是否表示死锁
403///
404/// 各数据库死锁错误码：
405/// - MySQL: 1213 (ER_LOCK_DEADLOCK)
406/// - PostgreSQL: 40P01 (deadlock_detected)
407/// - SQLite: "database is locked" / "database table is locked"
408/// - Oracle: ORA-00060
409/// - SQL Server: 1205 (Transaction was deadlocked)
410pub fn is_deadlock_error(err_msg: &str) -> bool {
411    let lower = err_msg.to_lowercase();
412    // MySQL: "Deadlock found when trying to get lock" (error 1213)
413    if lower.contains("deadlock found when trying to get lock") {
414        return true;
415    }
416    // MySQL error code 1213
417    if lower.contains("error 1213") || lower.contains("(1213)") {
418        return true;
419    }
420    // PostgreSQL: "deadlock detected" (SQLSTATE 40P01)
421    if lower.contains("deadlock detected") || lower.contains("40p01") {
422        return true;
423    }
424    // SQLite: "database is locked"
425    if lower.contains("database is locked") || lower.contains("database table is locked") {
426        return true;
427    }
428    // Oracle: ORA-00060: deadlock detected while waiting for resource
429    if lower.contains("ora-00060") {
430        return true;
431    }
432    // SQL Server: 1205 (Transaction was deadlocked on lock resources)
433    if lower.contains("transaction (process id") && lower.contains("was deadlocked") {
434        return true;
435    }
436    if lower.contains("error 1205") || lower.contains("(1205)") {
437        return true;
438    }
439    false
440}
441
442/// M-8 修复：在死锁时自动重试事务
443///
444/// `operation` 是一个异步闭包，返回 `Result<T, TxError>`。如果返回的错误包含
445/// 死锁信息（通过 `is_deadlock_error` 判断），则等待 `backoff` 后重试。
446///
447/// # 参数
448///
449/// - `max_attempts`: 最大重试次数（含首次执行），默认 3
450/// - `backoff`: 每次重试前的等待时间，默认 50ms
451/// - `operation`: 异步闭包，接受当前尝试次数（从 1 开始），返回 `Result<T, TxError>`
452///
453/// # 返回值
454///
455/// - 成功时返回 `Ok(T)`
456/// - 所有重试都失败时返回最后一次的 `Err(TxError::DeadlockDetected)`
457///
458/// # 示例
459///
460/// ```ignore
461/// let result = retry_on_deadlock(3, Duration::from_millis(50), |attempt| async move {
462///     // 执行事务操作
463///     tx.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1").await
464/// }).await;
465/// ```
466pub async fn retry_on_deadlock<F, Fut, T>(
467    max_attempts: u32,
468    backoff: Duration,
469    operation: F,
470) -> Result<T, TxError>
471where
472    F: Fn(u32) -> Fut,
473    Fut: std::future::Future<Output = Result<T, TxError>>,
474{
475    let mut last_err: Option<TxError> = None;
476    for attempt in 1..=max_attempts {
477        match operation(attempt).await {
478            Ok(v) => return Ok(v),
479            Err(e) => {
480                // 检查是否为死锁错误
481                let err_msg = format!("{}", e);
482                if is_deadlock_error(&err_msg) && attempt < max_attempts {
483                    tokio::time::sleep(backoff).await;
484                    last_err = Some(TxError::DeadlockDetected {
485                        attempt,
486                        max_attempts,
487                    });
488                    continue;
489                }
490                // 非死锁错误或已达最大重试次数，直接返回
491                return Err(e);
492            }
493        }
494    }
495    // 理论上不会到达（循环内所有路径都会 return），但为类型安全保留
496    Err(last_err.unwrap_or(TxError::DeadlockDetected {
497        attempt: max_attempts,
498        max_attempts,
499    }))
500}
501
502impl Drop for Transaction {
503    fn drop(&mut self) {
504        // 如果事务未被显式提交或回滚，在 drop 时尝试回滚
505        // 注意：无法在 Drop 中 await，所以这里 spawn 一个后台任务执行 rollback
506        if self.state == TransactionState::Active {
507            let conn = self.conn.clone();
508            // 尝试获取当前 tokio 运行时句柄；若不存在（如非 async 上下文）则跳过
509            if let Ok(handle) = tokio::runtime::Handle::try_current() {
510                // spawn 后台任务：锁连接 → rollback → 连接随 Arc 释放而 Drop
511                // 若任务因 runtime 关闭未执行，连接 Drop 时由驱动/池策略兜底
512                handle.spawn(async move {
513                    let mut conn_guard = conn.lock().await;
514                    if let Some(ref mut conn) = *conn_guard {
515                        let _ = conn.rollback().await;
516                    }
517                });
518            }
519            // 标记为已回滚（即使后台任务未完成，状态机也需要前进）
520            self.state = TransactionState::RolledBack;
521        }
522    }
523}
524
525/// 事务管理器，管理多个事务
526/// 事务管理器（按名称管理多个事务）
527pub struct TransactionManager {
528    transactions: Arc<Mutex<std::collections::HashMap<String, Transaction>>>,
529}
530
531impl TransactionManager {
532    /// 创建空的事务管理器
533    pub fn new() -> Self {
534        Self {
535            transactions: Arc::new(Mutex::new(std::collections::HashMap::new())),
536        }
537    }
538
539    /// 开始新事务
540    pub async fn begin(
541        &self,
542        id: String,
543        conn: Box<dyn Connection>,
544        options: TransactOptions,
545    ) -> Result<(), TxError> {
546        let mut conn = conn;
547        conn.begin_transaction()
548            .await
549            .map_err(|e| TxError::CommitFailed(e.to_string()))?;
550        let tx = Transaction::new(conn, options);
551        let mut txs = self.transactions.lock().await;
552        txs.insert(id, tx);
553        Ok(())
554    }
555
556    /// 提交事务
557    pub async fn commit(&self, id: &str) -> Result<(), TxError> {
558        let mut txs = self.transactions.lock().await;
559        let tx = txs
560            .get_mut(id)
561            .ok_or_else(|| TxError::SavepointError(format!("Transaction {} not found", id)))?;
562        tx.commit().await
563    }
564
565    /// 回滚事务
566    pub async fn rollback(&self, id: &str) -> Result<(), TxError> {
567        let mut txs = self.transactions.lock().await;
568        let tx = txs
569            .get_mut(id)
570            .ok_or_else(|| TxError::SavepointError(format!("Transaction {} not found", id)))?;
571        tx.rollback().await
572    }
573
574    /// 获取事务状态
575    pub async fn state(&self, id: &str) -> Option<TransactionState> {
576        let txs = self.transactions.lock().await;
577        txs.get(id).map(|tx| tx.state())
578    }
579
580    /// 列出所有事务 ID
581    pub async fn list(&self) -> Vec<String> {
582        let txs = self.transactions.lock().await;
583        txs.keys().cloned().collect()
584    }
585
586    /// 移除已完成的事务
587    pub async fn remove(&self, id: &str) -> Option<Transaction> {
588        let mut txs = self.transactions.lock().await;
589        txs.remove(id)
590    }
591}
592
593impl Default for TransactionManager {
594    fn default() -> Self {
595        Self::new()
596    }
597}
598
599#[cfg(test)]
600mod tests {
601    use super::*;
602    use std::future::Future;
603    use std::pin::Pin;
604
605    /// 测试用的模拟连接
606    struct MockConnection {
607        begin_called: bool,
608        commit_called: bool,
609        rollback_called: bool,
610        executed_sql: Vec<String>,
611    }
612
613    impl MockConnection {
614        fn new() -> Self {
615            Self {
616                begin_called: false,
617                commit_called: false,
618                rollback_called: false,
619                executed_sql: Vec::new(),
620            }
621        }
622    }
623
624    impl Connection for MockConnection {
625        fn execute<'a>(
626            &'a mut self,
627            sql: &'a str,
628        ) -> Pin<Box<dyn Future<Output = Result<u64, crate::DbError>> + Send + 'a>> {
629            Box::pin(async move {
630                self.executed_sql.push(sql.to_string());
631                Ok(1)
632            })
633        }
634
635        fn query<'a>(
636            &'a mut self,
637            _sql: &'a str,
638        ) -> Pin<
639            Box<
640                dyn Future<
641                        Output = Result<
642                            Vec<std::collections::HashMap<String, crate::value::Value>>,
643                            crate::DbError,
644                        >,
645                    > + Send
646                    + 'a,
647            >,
648        > {
649            Box::pin(async move { Ok(vec![]) })
650        }
651
652        fn begin_transaction<'a>(
653            &'a mut self,
654        ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
655            Box::pin(async move {
656                self.begin_called = true;
657                Ok(())
658            })
659        }
660
661        fn commit<'a>(
662            &'a mut self,
663        ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
664            Box::pin(async move {
665                self.commit_called = true;
666                Ok(())
667            })
668        }
669
670        fn rollback<'a>(
671            &'a mut self,
672        ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
673            Box::pin(async move {
674                self.rollback_called = true;
675                Ok(())
676            })
677        }
678
679        fn is_connected(&self) -> bool {
680            true
681        }
682
683        fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
684            Box::pin(async move { true })
685        }
686
687        fn close<'a>(
688            &'a mut self,
689        ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
690            Box::pin(async move { Ok(()) })
691        }
692    }
693
694    #[test]
695    fn test_isolation_level_display() {
696        assert_eq!(IsolationLevel::ReadCommitted.to_string(), "READ COMMITTED");
697        assert_eq!(IsolationLevel::Serializable.to_string(), "SERIALIZABLE");
698    }
699
700    #[test]
701    fn test_transaction_state_default() {
702        let opts = TransactOptions::default();
703        assert!(opts.isolation_level.is_none());
704        assert!(!opts.read_only);
705    }
706
707    #[test]
708    fn test_transact_options_builder() {
709        let opts = TransactOptions {
710            isolation_level: Some(IsolationLevel::Serializable),
711            read_only: true,
712            timeout: Some(Duration::from_secs(30)),
713            max_nesting_depth: DEFAULT_MAX_NESTING_DEPTH,
714            propagation: PropagationBehavior::default(),
715        };
716
717        assert_eq!(opts.isolation_level, Some(IsolationLevel::Serializable));
718        assert!(opts.read_only);
719        assert_eq!(opts.timeout, Some(Duration::from_secs(30)));
720    }
721
722    #[test]
723    fn test_auto_commit_default() {
724        assert_eq!(AutoCommit::default(), AutoCommit::On);
725    }
726
727    #[test]
728    fn test_transaction_state() {
729        assert_eq!(TransactionState::Active, TransactionState::Active);
730        assert_ne!(TransactionState::Active, TransactionState::Committed);
731    }
732
733    #[test]
734    fn test_transact_options_chaining() {
735        let opts = TransactOptions::default()
736            .with_isolation(IsolationLevel::Serializable)
737            .read_only()
738            .with_timeout(Duration::from_secs(60));
739        assert_eq!(opts.isolation_level, Some(IsolationLevel::Serializable));
740        assert!(opts.read_only);
741        assert_eq!(opts.timeout, Some(Duration::from_secs(60)));
742    }
743
744    #[tokio::test]
745    async fn test_transaction_commit() -> Result<(), TxError> {
746        let conn = Box::new(MockConnection::new());
747        let mut tx = Transaction::new(conn, TransactOptions::default());
748        assert!(tx.is_active());
749
750        let result = tx.execute("INSERT INTO users VALUES (1)").await;
751        assert!(result.is_ok());
752
753        tx.commit().await?;
754        assert_eq!(tx.state(), TransactionState::Committed);
755
756        // 再次 commit 应该失败（NotActive）
757        let result = tx.commit().await;
758        assert!(result.is_err());
759        match result {
760            Err(TxError::NotActive(state)) => {
761                assert_eq!(state, TransactionState::Committed);
762            }
763            _ => panic!("Expected NotActive error"),
764        }
765        Ok(())
766    }
767
768    #[tokio::test]
769    async fn test_transaction_rollback() -> Result<(), TxError> {
770        let conn = Box::new(MockConnection::new());
771        let mut tx = Transaction::new(conn, TransactOptions::default());
772
773        tx.rollback().await?;
774        assert_eq!(tx.state(), TransactionState::RolledBack);
775
776        // 再次 rollback 应该失败（NotActive）
777        let result = tx.rollback().await;
778        assert!(result.is_err());
779        match result {
780            Err(TxError::NotActive(state)) => {
781                assert_eq!(state, TransactionState::RolledBack);
782            }
783            _ => panic!("Expected NotActive error"),
784        }
785        Ok(())
786    }
787
788    #[tokio::test]
789    async fn test_transaction_execute_after_commit() -> Result<(), TxError> {
790        let conn = Box::new(MockConnection::new());
791        let mut tx = Transaction::new(conn, TransactOptions::default());
792        tx.commit().await?;
793
794        let result = tx.execute("SELECT 1").await;
795        assert!(result.is_err());
796        match result {
797            Err(TxError::NotActive(_)) => {}
798            _ => panic!("Expected NotActive error"),
799        }
800        Ok(())
801    }
802
803    #[tokio::test]
804    async fn test_transaction_query_after_commit_returns_not_active() -> Result<(), TxError> {
805        let conn = Box::new(MockConnection::new());
806        let mut tx = Transaction::new(conn, TransactOptions::default());
807        tx.commit().await?;
808
809        let result = tx.query("SELECT 1").await;
810        assert!(result.is_err());
811        match result {
812            Err(TxError::NotActive(_)) => {}
813            _ => panic!("Expected NotActive error"),
814        }
815        Ok(())
816    }
817
818    #[tokio::test]
819    async fn test_transaction_savepoint() -> Result<(), TxError> {
820        let conn = Box::new(MockConnection::new());
821        let mut tx = Transaction::new(conn, TransactOptions::default());
822
823        let sp1 = tx.savepoint().await?;
824        assert_eq!(sp1, "sp_1");
825
826        let sp2 = tx.savepoint().await?;
827        assert_eq!(sp2, "sp_2");
828
829        tx.rollback_to_savepoint(&sp1).await?;
830        tx.release_savepoint(&sp2).await?;
831        Ok(())
832    }
833
834    #[tokio::test]
835    async fn test_transaction_savepoint_name_validation() {
836        let conn = Box::new(MockConnection::new());
837        let mut tx = Transaction::new(conn, TransactOptions::default());
838
839        // 非法名称：包含单引号（SQL 注入尝试）
840        let result = tx.rollback_to_savepoint("sp'; DROP TABLE--").await;
841        assert!(result.is_err());
842        match result {
843            Err(TxError::InvalidSavepointName(_)) => {}
844            _ => panic!("Expected InvalidSavepointName error"),
845        }
846
847        // 非法名称：以数字开头
848        let result = tx.release_savepoint("1sp").await;
849        assert!(result.is_err());
850        match result {
851            Err(TxError::InvalidSavepointName(_)) => {}
852            _ => panic!("Expected InvalidSavepointName error"),
853        }
854
855        // 非法名称：空字符串
856        let result = tx.rollback_to_savepoint("").await;
857        assert!(result.is_err());
858        match result {
859            Err(TxError::InvalidSavepointName(_)) => {}
860            _ => panic!("Expected InvalidSavepointName error"),
861        }
862
863        // 合法名称：字母+下划线+数字
864        let result = tx.rollback_to_savepoint("sp_test_1").await;
865        assert!(result.is_ok());
866    }
867
868    #[tokio::test]
869    async fn test_transaction_take_connection() -> Result<(), TxError> {
870        let conn = Box::new(MockConnection::new());
871        let mut tx = Transaction::new(conn, TransactOptions::default());
872
873        // Active 状态下不能取连接
874        let result = tx.take_connection().await;
875        assert!(result.is_err());
876        match result {
877            Err(TxError::NotActive(_)) => {}
878            _ => panic!("Expected NotActive error"),
879        }
880
881        // commit 后可以取连接
882        tx.commit().await?;
883        let conn = tx.take_connection().await;
884        assert!(conn.is_ok());
885
886        // 重复取连接应失败
887        let result = tx.take_connection().await;
888        assert!(result.is_err());
889        match result {
890            Err(TxError::ConnectionTaken) => {}
891            _ => panic!("Expected ConnectionTaken error"),
892        }
893        Ok(())
894    }
895
896    #[tokio::test]
897    async fn test_transaction_manager() -> Result<(), TxError> {
898        let mgr = TransactionManager::new();
899        let conn = Box::new(MockConnection::new());
900
901        mgr.begin("tx1".to_string(), conn, TransactOptions::default())
902            .await?;
903
904        let state = mgr.state("tx1").await;
905        assert_eq!(state, Some(TransactionState::Active));
906
907        mgr.commit("tx1").await?;
908        let state = mgr.state("tx1").await;
909        assert_eq!(state, Some(TransactionState::Committed));
910
911        let list = mgr.list().await;
912        assert!(list.contains(&"tx1".to_string()));
913        Ok(())
914    }
915
916    #[tokio::test]
917    async fn test_transaction_manager_rollback() -> Result<(), TxError> {
918        let mgr = TransactionManager::new();
919        let conn = Box::new(MockConnection::new());
920
921        mgr.begin("tx2".to_string(), conn, TransactOptions::default())
922            .await?;
923
924        mgr.rollback("tx2").await?;
925        let state = mgr.state("tx2").await;
926        assert_eq!(state, Some(TransactionState::RolledBack));
927        Ok(())
928    }
929
930    #[tokio::test]
931    async fn test_transaction_manager_not_found() {
932        let mgr = TransactionManager::new();
933        let result = mgr.commit("nonexistent").await;
934        assert!(result.is_err());
935    }
936
937    #[tokio::test]
938    async fn test_transaction_manager_remove() -> Result<(), TxError> {
939        let mgr = TransactionManager::new();
940        let conn = Box::new(MockConnection::new());
941
942        mgr.begin("tx3".to_string(), conn, TransactOptions::default())
943            .await?;
944
945        let removed = mgr.remove("tx3").await;
946        assert!(removed.is_some());
947
948        let state = mgr.state("tx3").await;
949        assert_eq!(state, None);
950        Ok(())
951    }
952
953    /// 验证 Drop 时若事务仍 Active，会 spawn 后台 rollback 任务
954    #[tokio::test]
955    async fn test_transaction_drop_rolls_back_when_active() {
956        use std::sync::atomic::{AtomicBool, Ordering};
957        use std::sync::Arc as StdArc;
958
959        struct TrackingConnection {
960            rollback_called: StdArc<AtomicBool>,
961        }
962
963        impl Connection for TrackingConnection {
964            fn execute<'a>(
965                &'a mut self,
966                _sql: &'a str,
967            ) -> Pin<Box<dyn Future<Output = Result<u64, crate::DbError>> + Send + 'a>>
968            {
969                Box::pin(async { Ok(1) })
970            }
971            fn query<'a>(
972                &'a mut self,
973                _sql: &'a str,
974            ) -> Pin<
975                Box<
976                    dyn Future<
977                            Output = Result<
978                                Vec<std::collections::HashMap<String, crate::value::Value>>,
979                                crate::DbError,
980                            >,
981                        > + Send
982                        + 'a,
983                >,
984            > {
985                Box::pin(async { Ok(vec![]) })
986            }
987            fn begin_transaction<'a>(
988                &'a mut self,
989            ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
990                Box::pin(async { Ok(()) })
991            }
992            fn commit<'a>(
993                &'a mut self,
994            ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
995                Box::pin(async { Ok(()) })
996            }
997            fn rollback<'a>(
998                &'a mut self,
999            ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
1000                let flag = self.rollback_called.clone();
1001                Box::pin(async move {
1002                    flag.store(true, Ordering::SeqCst);
1003                    Ok(())
1004                })
1005            }
1006            fn is_connected(&self) -> bool {
1007                true
1008            }
1009            fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
1010                Box::pin(async { true })
1011            }
1012            fn close<'a>(
1013                &'a mut self,
1014            ) -> Pin<Box<dyn Future<Output = Result<(), crate::DbError>> + Send + 'a>> {
1015                Box::pin(async { Ok(()) })
1016            }
1017        }
1018
1019        let rollback_flag = StdArc::new(AtomicBool::new(false));
1020        let conn = Box::new(TrackingConnection {
1021            rollback_called: rollback_flag.clone(),
1022        });
1023        {
1024            let _tx = Transaction::new(conn, TransactOptions::default());
1025            // _tx 在块结束时 drop，状态为 Active，应触发后台 rollback
1026        }
1027        // 给 spawn 的任务一点时间执行
1028        tokio::time::sleep(Duration::from_millis(50)).await;
1029        assert!(
1030            rollback_flag.load(Ordering::SeqCst),
1031            "Drop should have triggered rollback"
1032        );
1033    }
1034
1035    // ==================== H-8 事务嵌套深度限制测试 ====================
1036
1037    #[test]
1038    fn test_h8_default_max_nesting_depth_is_8() {
1039        let opts = TransactOptions::default();
1040        assert_eq!(opts.max_nesting_depth, DEFAULT_MAX_NESTING_DEPTH);
1041        assert_eq!(opts.max_nesting_depth, 8);
1042    }
1043
1044    #[test]
1045    fn test_h8_with_max_nesting_depth_builder() {
1046        let opts = TransactOptions::default().with_max_nesting_depth(3);
1047        assert_eq!(opts.max_nesting_depth, 3);
1048    }
1049
1050    #[tokio::test]
1051    async fn test_h8_savepoint_within_default_depth_succeeds() -> Result<(), TxError> {
1052        let conn = Box::new(MockConnection::new());
1053        let mut tx = Transaction::new(conn, TransactOptions::default());
1054
1055        // 默认深度 8，创建 8 个保存点应全部成功
1056        for i in 1..=8 {
1057            let sp = tx.savepoint().await?;
1058            assert_eq!(sp, format!("sp_{}", i));
1059        }
1060        Ok(())
1061    }
1062
1063    #[tokio::test]
1064    async fn test_h8_savepoint_exceeding_default_depth_fails() -> Result<(), TxError> {
1065        let conn = Box::new(MockConnection::new());
1066        let mut tx = Transaction::new(conn, TransactOptions::default());
1067
1068        // 创建 8 个保存点（达到上限）
1069        for _ in 0..8 {
1070            tx.savepoint().await?;
1071        }
1072
1073        // 第 9 个保存点应失败
1074        let result = tx.savepoint().await;
1075        assert!(result.is_err());
1076        match result {
1077            Err(TxError::MaxNestingDepthExceeded {
1078                current_depth,
1079                max_depth,
1080            }) => {
1081                assert_eq!(current_depth, 9);
1082                assert_eq!(max_depth, 8);
1083            }
1084            _ => panic!("Expected MaxNestingDepthExceeded error"),
1085        }
1086        Ok(())
1087    }
1088
1089    #[tokio::test]
1090    async fn test_h8_savepoint_with_custom_depth_3() -> Result<(), TxError> {
1091        let conn = Box::new(MockConnection::new());
1092        let mut tx = Transaction::new(conn, TransactOptions::default().with_max_nesting_depth(3));
1093
1094        // 3 个保存点应成功
1095        for i in 1..=3 {
1096            let sp = tx.savepoint().await?;
1097            assert_eq!(sp, format!("sp_{}", i));
1098        }
1099
1100        // 第 4 个应失败
1101        let result = tx.savepoint().await;
1102        assert!(result.is_err());
1103        match result {
1104            Err(TxError::MaxNestingDepthExceeded {
1105                current_depth,
1106                max_depth,
1107            }) => {
1108                assert_eq!(current_depth, 4);
1109                assert_eq!(max_depth, 3);
1110            }
1111            _ => panic!("Expected MaxNestingDepthExceeded error"),
1112        }
1113        Ok(())
1114    }
1115
1116    #[tokio::test]
1117    async fn test_h8_savepoint_depth_zero_disables_nesting() {
1118        let conn = Box::new(MockConnection::new());
1119        let mut tx = Transaction::new(conn, TransactOptions::default().with_max_nesting_depth(0));
1120
1121        // 深度 0 表示禁用嵌套，首次 savepoint 即失败
1122        let result = tx.savepoint().await;
1123        assert!(result.is_err());
1124        match result {
1125            Err(TxError::MaxNestingDepthExceeded {
1126                current_depth,
1127                max_depth,
1128            }) => {
1129                assert_eq!(current_depth, 1);
1130                assert_eq!(max_depth, 0);
1131            }
1132            _ => panic!("Expected MaxNestingDepthExceeded error"),
1133        }
1134    }
1135
1136    #[tokio::test]
1137    async fn test_h8_savepoint_after_rollback_to_still_respects_depth() -> Result<(), TxError> {
1138        // 即使回滚到保存点，savepoint_counter 不减少（保存点栈可能仍存在），
1139        // 因此深度检查仍以 savepoint_counter 为准
1140        let conn = Box::new(MockConnection::new());
1141        let mut tx = Transaction::new(conn, TransactOptions::default().with_max_nesting_depth(2));
1142
1143        let sp1 = tx.savepoint().await?;
1144        let sp2 = tx.savepoint().await?;
1145
1146        // 回滚到 sp1（不重置计数器）
1147        tx.rollback_to_savepoint(&sp1).await?;
1148        tx.release_savepoint(&sp2).await?;
1149
1150        // 第 3 个保存点仍应失败（计数器不回退）
1151        let result = tx.savepoint().await;
1152        assert!(result.is_err());
1153        match result {
1154            Err(TxError::MaxNestingDepthExceeded {
1155                current_depth,
1156                max_depth,
1157            }) => {
1158                assert_eq!(current_depth, 3);
1159                assert_eq!(max_depth, 2);
1160            }
1161            _ => panic!("Expected MaxNestingDepthExceeded error"),
1162        }
1163        Ok(())
1164    }
1165
1166    #[tokio::test]
1167    async fn test_h8_max_nesting_depth_error_display() {
1168        let err = TxError::MaxNestingDepthExceeded {
1169            current_depth: 10,
1170            max_depth: 8,
1171        };
1172        let msg = format!("{}", err);
1173        assert!(msg.contains("10"));
1174        assert!(msg.contains("8"));
1175        assert!(msg.contains("exceeds"));
1176    }
1177
1178    // ==================== M-8 死锁检测重试测试 ====================
1179
1180    #[test]
1181    fn test_m8_is_deadlock_error_mysql() {
1182        assert!(is_deadlock_error(
1183            "Deadlock found when trying to get lock; try restarting transaction"
1184        ));
1185        assert!(is_deadlock_error("Error 1213: Deadlock found"));
1186        assert!(is_deadlock_error("MySQL error (1213)"));
1187    }
1188
1189    #[test]
1190    fn test_m8_is_deadlock_error_postgresql() {
1191        assert!(is_deadlock_error("deadlock detected"));
1192        assert!(is_deadlock_error("ERROR: deadlock detected (40P01)"));
1193        assert!(is_deadlock_error("SQLSTATE 40P01"));
1194    }
1195
1196    #[test]
1197    fn test_m8_is_deadlock_error_sqlite() {
1198        assert!(is_deadlock_error("database is locked"));
1199        assert!(is_deadlock_error("database table is locked"));
1200    }
1201
1202    #[test]
1203    fn test_m8_is_deadlock_error_oracle() {
1204        assert!(is_deadlock_error(
1205            "ORA-00060: deadlock detected while waiting for resource"
1206        ));
1207    }
1208
1209    #[test]
1210    fn test_m8_is_deadlock_error_sql_server() {
1211        assert!(is_deadlock_error(
1212            "Transaction (Process ID 52) was deadlocked on lock resources"
1213        ));
1214        assert!(is_deadlock_error("Error 1205: Transaction was deadlocked"));
1215    }
1216
1217    #[test]
1218    fn test_m8_is_deadlock_error_non_deadlock() {
1219        assert!(!is_deadlock_error("connection refused"));
1220        assert!(!is_deadlock_error("syntax error near SELECT"));
1221        assert!(!is_deadlock_error("permission denied for table users"));
1222        assert!(!is_deadlock_error(""));
1223    }
1224
1225    #[tokio::test]
1226    async fn test_m8_retry_on_deadlock_succeeds_first_attempt() -> Result<(), TxError> {
1227        use std::sync::atomic::{AtomicU32, Ordering};
1228
1229        let counter = Arc::new(AtomicU32::new(0));
1230        let counter_clone = counter.clone();
1231
1232        let result: Result<u32, TxError> =
1233            retry_on_deadlock(3, Duration::from_millis(1), |_attempt| {
1234                let c = counter_clone.clone();
1235                async move {
1236                    c.fetch_add(1, Ordering::SeqCst);
1237                    Ok(42u32)
1238                }
1239            })
1240            .await;
1241
1242        assert_eq!(result?, 42);
1243        assert_eq!(counter.load(Ordering::SeqCst), 1);
1244        Ok(())
1245    }
1246
1247    #[tokio::test]
1248    async fn test_m8_retry_on_deadlock_retries_on_deadlock_error() -> Result<(), TxError> {
1249        use std::sync::atomic::{AtomicU32, Ordering};
1250
1251        let counter = Arc::new(AtomicU32::new(0));
1252        let counter_clone = counter.clone();
1253
1254        let result: Result<u32, TxError> =
1255            retry_on_deadlock(3, Duration::from_millis(1), |_attempt| {
1256                let c = counter_clone.clone();
1257                async move {
1258                    let n = c.fetch_add(1, Ordering::SeqCst);
1259                    if n < 2 {
1260                        // 前两次返回死锁错误
1261                        Err(TxError::CommitFailed(
1262                            "Deadlock found when trying to get lock".to_string(),
1263                        ))
1264                    } else {
1265                        Ok(42u32)
1266                    }
1267                }
1268            })
1269            .await;
1270
1271        assert_eq!(result?, 42);
1272        assert_eq!(counter.load(Ordering::SeqCst), 3);
1273        Ok(())
1274    }
1275
1276    #[tokio::test]
1277    async fn test_m8_retry_on_deadlock_returns_error_after_max_attempts() {
1278        use std::sync::atomic::{AtomicU32, Ordering};
1279
1280        let counter = Arc::new(AtomicU32::new(0));
1281        let counter_clone = counter.clone();
1282
1283        let result: Result<u32, TxError> =
1284            retry_on_deadlock(2, Duration::from_millis(1), |_attempt| {
1285                let c = counter_clone.clone();
1286                async move {
1287                    c.fetch_add(1, Ordering::SeqCst);
1288                    Err(TxError::CommitFailed(
1289                        "Deadlock found when trying to get lock".to_string(),
1290                    ))
1291                }
1292            })
1293            .await;
1294
1295        // 应返回最后一次的死锁错误
1296        assert!(result.is_err());
1297        assert_eq!(counter.load(Ordering::SeqCst), 2);
1298    }
1299
1300    #[tokio::test]
1301    async fn test_m8_retry_on_deadlock_does_not_retry_non_deadlock_errors() {
1302        use std::sync::atomic::{AtomicU32, Ordering};
1303
1304        let counter = Arc::new(AtomicU32::new(0));
1305        let counter_clone = counter.clone();
1306
1307        let result: Result<u32, TxError> =
1308            retry_on_deadlock(3, Duration::from_millis(1), |_attempt| {
1309                let c = counter_clone.clone();
1310                async move {
1311                    c.fetch_add(1, Ordering::SeqCst);
1312                    Err(TxError::CommitFailed("syntax error".to_string()))
1313                }
1314            })
1315            .await;
1316
1317        // 非死锁错误应立即返回，不重试
1318        assert!(result.is_err());
1319        assert_eq!(counter.load(Ordering::SeqCst), 1);
1320    }
1321
1322    #[test]
1323    fn test_m8_deadlock_error_display() {
1324        let err = TxError::DeadlockDetected {
1325            attempt: 2,
1326            max_attempts: 3,
1327        };
1328        let msg = format!("{}", err);
1329        assert!(msg.contains("2"));
1330        assert!(msg.contains("3"));
1331        assert!(msg.contains("Deadlock"));
1332    }
1333}
sz_orm_core/transaction.rs

sz_orm_core/
transaction.rs
