aster_forge_db/
transaction.rs

1//! Transaction helpers with consistent error mapping and rollback tracing.
2//!
3//! Service code can either manually begin/commit/rollback with uniform error conversion or run a
4//! callback inside `with_transaction`. The rollback guard logs dropped transactions, which helps
5//! catch early-return paths that rely on rollback-on-drop instead of making rollback explicit.
6//!
7//! ## Pattern
8//!
9//! Standard service-layer transaction usage:
10//! ```ignore
11//! transaction::with_transaction(db, async |txn| {
12//!     repo::operation(txn, ...).await?;
13//!     repo::another_operation(txn, ...).await?;
14//!     Ok(())
15//! })
16//! .await?;
17//! ```
18
19use crate::{DbError, Result, database_error_kind};
20use sea_orm::TransactionSession;
21use std::panic::Location;
22use std::{future::Future, ops::AsyncFnOnce, pin::Pin, time::Duration};
23
24struct RollbackGuard {
25    file: &'static str,
26    line: u32,
27    armed: bool,
28}
29
30impl RollbackGuard {
31    fn new(location: &'static Location<'static>) -> Self {
32        Self {
33            file: location.file(),
34            line: location.line(),
35            armed: true,
36        }
37    }
38
39    fn disarm(&mut self) {
40        self.armed = false;
41    }
42}
43
44impl Drop for RollbackGuard {
45    fn drop(&mut self) {
46        if self.armed {
47            tracing::warn!(
48                file = self.file,
49                line = self.line,
50                "transaction dropped before explicit commit/rollback; relying on rollback-on-drop"
51            );
52        }
53    }
54}
55
56/// Begins and returns a transaction so the caller can commit or roll it back.
57///
58/// This centralizes `begin` error mapping.
59///
60/// # Errors
61///
62/// Returns an error when transaction setup, execution, or finalization fails.
63pub async fn begin<C: sea_orm::TransactionTrait>(db: &C) -> Result<C::Transaction> {
64    db.begin()
65        .await
66        .map_err(|error| database_operation_with_context(&error, "begin transaction"))
67}
68
69/// Commits a transaction and maps errors consistently.
70///
71/// # Errors
72///
73/// Returns an error when transaction setup, execution, or finalization fails.
74pub async fn commit<T: sea_orm::TransactionSession>(txn: T) -> Result<()> {
75    txn.commit()
76        .await
77        .map_err(|error| database_operation_with_context(&error, "commit transaction"))
78}
79
80/// Rolls back a transaction and maps errors consistently.
81///
82/// # Errors
83///
84/// Returns an error when transaction setup, execution, or finalization fails.
85pub async fn rollback<T: sea_orm::TransactionSession>(txn: T) -> Result<()> {
86    txn.rollback()
87        .await
88        .map_err(|error| database_operation_with_context(&error, "rollback transaction"))
89}
90
91fn database_operation_with_context(error: &sea_orm::DbErr, context: &str) -> DbError {
92    let kind = database_error_kind(error);
93    let message = format!("{context}: {error}");
94    match kind {
95        Some(kind) => DbError::database_operation_classified(message, kind),
96        None => DbError::database_operation(message),
97    }
98}
99
100fn transaction_delay(config: &crate::retry::RetryConfig, attempt: u32) -> Duration {
101    aster_forge_utils::backoff::cap_delay(
102        aster_forge_utils::backoff::exponential_delay(
103            Duration::from_millis(config.base_delay_ms),
104            attempt,
105        ),
106        Duration::from_millis(config.max_delay_ms),
107    )
108}
109
110#[derive(Clone, Copy, Debug, Eq, PartialEq)]
111enum CommitFailureAction {
112    Retry,
113    ReturnKnownFailure,
114    ReturnOutcomeUnknown,
115}
116
117fn commit_failure_action(
118    retryable: bool,
119    outcome_known_rolled_back: bool,
120    attempt: u32,
121    max_retries: u32,
122) -> CommitFailureAction {
123    if !outcome_known_rolled_back {
124        CommitFailureAction::ReturnOutcomeUnknown
125    } else if retryable {
126        if attempt < max_retries {
127            CommitFailureAction::Retry
128        } else {
129            CommitFailureAction::ReturnKnownFailure
130        }
131    } else {
132        CommitFailureAction::ReturnKnownFailure
133    }
134}
135
136fn commit_outcome_known_rolled_back(kind: Option<crate::DatabaseErrorKind>) -> bool {
137    matches!(
138        kind,
139        Some(crate::DatabaseErrorKind::Deadlock | crate::DatabaseErrorKind::SerializationFailure)
140    )
141}
142
143/// Runs a complete transaction boundary with bounded retries selected by the product.
144///
145/// The callback is rerun only after rollback or a commit failure known to have rolled back the
146/// transaction. Any commit error with an uncertain server-side outcome is returned as
147/// `DbError::CommitOutcomeUnknown`, because the server may have committed the transaction even
148/// though the client did not receive a success response.
149///
150/// Use [`crate::retry::RetryConfig::deadlock`] as the config profile: deadlock and
151/// serialization conflicts resolve inside short lock-wait windows, so the fast profile fits.
152///
153/// # Errors
154///
155/// Returns an error when transaction setup, execution, or finalization fails.
156pub async fn with_transaction_retry<C, F, T, E, P>(
157    db: &C,
158    config: &crate::retry::RetryConfig,
159    mut operation: F,
160    should_retry: P,
161) -> std::result::Result<T, E>
162where
163    C: sea_orm::TransactionTrait,
164    F: for<'txn> FnMut(
165        &'txn C::Transaction,
166    )
167        -> Pin<Box<dyn Future<Output = std::result::Result<T, E>> + Send + 'txn>>,
168    E: From<DbError> + std::fmt::Display,
169    P: Fn(&E) -> bool,
170{
171    let mut attempt = 0;
172    loop {
173        let txn = match db.begin().await {
174            Ok(txn) => txn,
175            Err(error) => {
176                let error = E::from(database_operation_with_context(&error, "begin transaction"));
177                if attempt < config.max_retries && should_retry(&error) {
178                    tokio::time::sleep(transaction_delay(config, attempt)).await;
179                    attempt += 1;
180                    continue;
181                }
182                return Err(error);
183            }
184        };
185
186        match operation(&txn).await {
187            Ok(value) => match txn.commit().await {
188                Ok(()) => return Ok(value),
189                Err(error) => {
190                    let kind = database_error_kind(&error);
191                    let classified_error = E::from(match kind {
192                        Some(kind) => DbError::database_operation_classified(
193                            format!("commit transaction: {error}"),
194                            kind,
195                        ),
196                        None => DbError::database_operation(format!("commit transaction: {error}")),
197                    });
198                    match commit_failure_action(
199                        should_retry(&classified_error),
200                        commit_outcome_known_rolled_back(kind),
201                        attempt,
202                        config.max_retries,
203                    ) {
204                        CommitFailureAction::Retry => {
205                            tokio::time::sleep(transaction_delay(config, attempt)).await;
206                            attempt += 1;
207                        }
208                        CommitFailureAction::ReturnKnownFailure => return Err(classified_error),
209                        CommitFailureAction::ReturnOutcomeUnknown => {
210                            return Err(E::from(DbError::commit_outcome_unknown(
211                                format!("commit transaction: {error}"),
212                                kind,
213                            )));
214                        }
215                    }
216                }
217            },
218            Err(error) => {
219                if let Err(rollback_error) = txn.rollback().await {
220                    tracing::warn!(
221                        callback_error = %error,
222                        rollback_error = %rollback_error,
223                        "transaction rollback failed after callback error"
224                    );
225                }
226                if attempt < config.max_retries && should_retry(&error) {
227                    tokio::time::sleep(transaction_delay(config, attempt)).await;
228                    attempt += 1;
229                    continue;
230                }
231                return Err(error);
232            }
233        }
234    }
235}
236
237/// Runs a transaction callback with consistent tracing and rollback guarding.
238///
239/// The callback may return a product-specific error type. Forge-created transaction boundary
240/// errors are converted through `E: From<DbError>`, while callback errors are preserved unchanged.
241///
242/// # Errors
243///
244/// Returns an error when transaction setup, execution, or finalization fails.
245pub async fn with_transaction<C, F, T, E>(db: &C, operation: F) -> std::result::Result<T, E>
246where
247    C: sea_orm::TransactionTrait,
248    F: for<'txn> AsyncFnOnce(&'txn C::Transaction) -> std::result::Result<T, E>,
249    E: From<DbError> + std::fmt::Display,
250{
251    let location = Location::caller();
252    tracing::debug!(
253        file = location.file(),
254        line = location.line(),
255        "beginning transaction"
256    );
257    let txn = begin(db).await.map_err(E::from)?;
258    let mut rollback_guard = RollbackGuard::new(location);
259
260    match operation(&txn).await {
261        Ok(value) => {
262            rollback_guard.disarm();
263            commit(txn).await.map_err(E::from)?;
264            tracing::debug!(
265                file = location.file(),
266                line = location.line(),
267                "committed transaction"
268            );
269            Ok(value)
270        }
271        Err(error) => {
272            tracing::debug!(
273                file = location.file(),
274                line = location.line(),
275                error = %error,
276                "rolling back transaction after callback error"
277            );
278            rollback_guard.disarm();
279            if let Err(rollback_error) = rollback(txn).await {
280                tracing::error!(
281                    file = location.file(),
282                    line = location.line(),
283                    callback_error = %error,
284                    rollback_error = %rollback_error,
285                    "transaction rollback failed after callback error"
286                );
287            }
288            Err(error)
289        }
290    }
291}
292
293#[cfg(test)]
294mod tests {
295    use super::{
296        CommitFailureAction, commit_failure_action, commit_outcome_known_rolled_back, rollback,
297        transaction_delay, with_transaction,
298    };
299    use crate::{DbError, connection::DatabaseConfig};
300    use sea_orm::{ConnectionTrait, DatabaseConnection, Statement, TransactionTrait};
301    use std::fmt;
302
303    async fn sqlite_db() -> DatabaseConnection {
304        crate::connection::connect(&DatabaseConfig {
305            url: "sqlite::memory:".into(),
306            pool_size: 1,
307            retry_count: 0,
308        })
309        .await
310        .expect("sqlite memory database should connect")
311    }
312
313    async fn count_rows(db: &DatabaseConnection) -> i64 {
314        let statement = Statement::from_string(
315            sea_orm::DbBackend::Sqlite,
316            "SELECT COUNT(*) FROM transaction_items",
317        );
318        let row = db
319            .query_one_raw(statement)
320            .await
321            .expect("count query should succeed")
322            .expect("count query should return one row");
323        row.try_get_by_index(0).expect("count should decode")
324    }
325
326    #[derive(Debug, PartialEq, Eq)]
327    enum ProductError {
328        Db(String),
329        Validation(&'static str),
330    }
331
332    impl From<DbError> for ProductError {
333        fn from(value: DbError) -> Self {
334            Self::Db(value.to_string())
335        }
336    }
337
338    impl fmt::Display for ProductError {
339        fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
340            match self {
341                Self::Db(message) => formatter.write_str(message),
342                Self::Validation(message) => formatter.write_str(message),
343            }
344        }
345    }
346
347    #[tokio::test]
348    async fn with_transaction_commits_successful_callback() {
349        let db = sqlite_db().await;
350        db.execute_unprepared("CREATE TABLE transaction_items (id INTEGER PRIMARY KEY);")
351            .await
352            .expect("table should be created");
353
354        let value = with_transaction(&db, async |txn| {
355            txn.execute_unprepared("INSERT INTO transaction_items (id) VALUES (1);")
356                .await
357                .map_err(DbError::from)?;
358            Ok::<_, DbError>("committed")
359        })
360        .await
361        .expect("transaction should commit");
362
363        assert_eq!(value, "committed");
364        assert_eq!(count_rows(&db).await, 1);
365    }
366
367    #[tokio::test]
368    async fn with_transaction_rolls_back_callback_error() {
369        let db = sqlite_db().await;
370        db.execute_unprepared("CREATE TABLE transaction_items (id INTEGER PRIMARY KEY);")
371            .await
372            .expect("table should be created");
373
374        let error = with_transaction(&db, async |txn| {
375            txn.execute_unprepared("INSERT INTO transaction_items (id) VALUES (1);")
376                .await
377                .map_err(DbError::from)?;
378            Err::<(), _>(DbError::database_operation("forced failure"))
379        })
380        .await
381        .expect_err("callback error should propagate");
382
383        assert!(matches!(error, DbError::DatabaseOperation(_)));
384        assert_eq!(count_rows(&db).await, 0);
385    }
386
387    #[tokio::test]
388    async fn with_transaction_preserves_product_callback_errors() {
389        let db = sqlite_db().await;
390        db.execute_unprepared("CREATE TABLE transaction_items (id INTEGER PRIMARY KEY);")
391            .await
392            .expect("table should be created");
393
394        let error = with_transaction(&db, async |txn| {
395            txn.execute_unprepared("INSERT INTO transaction_items (id) VALUES (1);")
396                .await
397                .map_err(DbError::from)
398                .map_err(ProductError::from)?;
399            Err::<(), _>(ProductError::Validation("business validation failed"))
400        })
401        .await
402        .expect_err("callback error should propagate");
403
404        assert_eq!(
405            error,
406            ProductError::Validation("business validation failed")
407        );
408        assert_eq!(count_rows(&db).await, 0);
409    }
410
411    #[tokio::test]
412    async fn with_transaction_retry_restarts_after_callback_failure() {
413        use std::sync::{
414            Arc,
415            atomic::{AtomicUsize, Ordering},
416        };
417
418        let db = sqlite_db().await;
419        let attempts = Arc::new(AtomicUsize::new(0));
420        let config = crate::retry::RetryConfig {
421            max_retries: 2,
422            base_delay_ms: 0,
423            max_delay_ms: 0,
424        };
425        let result = {
426            let attempts = Arc::clone(&attempts);
427            super::with_transaction_retry(
428                &db,
429                &config,
430                move |_txn| {
431                    let attempts = Arc::clone(&attempts);
432                    Box::pin(async move {
433                        let attempt = attempts.fetch_add(1, Ordering::SeqCst);
434                        if attempt < 2 {
435                            Err(DbError::database_operation("temporary failure"))
436                        } else {
437                            Ok::<_, DbError>("ok")
438                        }
439                    })
440                },
441                |error| matches!(error, DbError::DatabaseOperation(_)),
442            )
443            .await
444        };
445
446        assert_eq!(result.unwrap(), "ok");
447        assert_eq!(attempts.load(Ordering::SeqCst), 3);
448    }
449
450    #[test]
451    fn transaction_delay_doubles_caps_and_handles_boundaries() {
452        let config = crate::retry::RetryConfig {
453            max_retries: 3,
454            base_delay_ms: 100,
455            max_delay_ms: 250,
456        };
457        assert_eq!(
458            transaction_delay(&config, 0),
459            std::time::Duration::from_millis(100)
460        );
461        assert_eq!(
462            transaction_delay(&config, 1),
463            std::time::Duration::from_millis(200)
464        );
465        assert_eq!(
466            transaction_delay(&config, 2),
467            std::time::Duration::from_millis(250)
468        );
469        assert_eq!(
470            transaction_delay(&config, u32::MAX),
471            std::time::Duration::from_millis(250)
472        );
473
474        let zero_max = crate::retry::RetryConfig {
475            max_retries: 1,
476            base_delay_ms: 100,
477            max_delay_ms: 0,
478        };
479        assert_eq!(transaction_delay(&zero_max, 0), std::time::Duration::ZERO);
480
481        let initial_above_max = crate::retry::RetryConfig {
482            max_retries: 1,
483            base_delay_ms: 1_000,
484            max_delay_ms: 250,
485        };
486        assert_eq!(
487            transaction_delay(&initial_above_max, 0),
488            std::time::Duration::from_millis(250)
489        );
490    }
491
492    #[tokio::test]
493    async fn with_transaction_retry_zero_budget_runs_callback_once() {
494        use std::sync::{
495            Arc,
496            atomic::{AtomicUsize, Ordering},
497        };
498
499        let db = sqlite_db().await;
500        let attempts = Arc::new(AtomicUsize::new(0));
501        let config = crate::retry::RetryConfig {
502            max_retries: 0,
503            base_delay_ms: 0,
504            max_delay_ms: 0,
505        };
506        let error = {
507            let attempts = attempts.clone();
508            super::with_transaction_retry(
509                &db,
510                &config,
511                move |_txn| {
512                    let attempts = attempts.clone();
513                    Box::pin(async move {
514                        attempts.fetch_add(1, Ordering::SeqCst);
515                        Err::<(), _>(DbError::database_operation("temporary failure"))
516                    })
517                },
518                |error| matches!(error, DbError::DatabaseOperation(_)),
519            )
520            .await
521            .unwrap_err()
522        };
523
524        assert!(matches!(error, DbError::DatabaseOperation(_)));
525        assert_eq!(attempts.load(Ordering::SeqCst), 1);
526    }
527
528    #[test]
529    fn commit_failure_action_preserves_known_exhausted_deadlocks() {
530        assert_eq!(
531            commit_failure_action(true, true, 0, 3),
532            CommitFailureAction::Retry
533        );
534        assert_eq!(
535            commit_failure_action(true, true, 3, 3),
536            CommitFailureAction::ReturnKnownFailure
537        );
538        assert_eq!(
539            commit_failure_action(false, false, 0, 3),
540            CommitFailureAction::ReturnOutcomeUnknown
541        );
542        assert_eq!(
543            commit_failure_action(true, false, 0, 3),
544            CommitFailureAction::ReturnOutcomeUnknown
545        );
546        assert_eq!(
547            commit_failure_action(false, true, 0, 3),
548            CommitFailureAction::ReturnKnownFailure
549        );
550        assert!(commit_outcome_known_rolled_back(Some(
551            crate::DatabaseErrorKind::Deadlock
552        )));
553        assert!(commit_outcome_known_rolled_back(Some(
554            crate::DatabaseErrorKind::SerializationFailure
555        )));
556        assert!(!commit_outcome_known_rolled_back(Some(
557            crate::DatabaseErrorKind::LockTimeout
558        )));
559    }
560
561    #[tokio::test]
562    async fn rollback_helper_discards_pending_changes() {
563        let db = sqlite_db().await;
564        db.execute_unprepared("CREATE TABLE transaction_items (id INTEGER PRIMARY KEY);")
565            .await
566            .expect("table should be created");
567        let txn = db.begin().await.expect("transaction should begin");
568        txn.execute_unprepared("INSERT INTO transaction_items (id) VALUES (1);")
569            .await
570            .expect("insert should succeed");
571
572        rollback(txn).await.expect("rollback should succeed");
573
574        assert_eq!(count_rows(&db).await, 0);
575    }
576}
aster_forge_db/transaction.rs

aster_forge_db/
transaction.rs
