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

sz_orm_core/
transaction.rs
