transaction.rs (2468B)
1 //! Transaction wrapper with OpenTelemetry instrumentation 2 //! 3 //! Provides a `with_transaction` helper that wraps database transactions 4 //! in a span that tracks commit/rollback outcomes. 5 6 use sqlx::{PgPool, Postgres, Transaction}; 7 use std::future::Future; 8 use std::pin::Pin; 9 use tracing::{instrument, Span}; 10 11 /// Execute a closure within a database transaction with OpenTelemetry instrumentation. 12 /// 13 /// This wrapper: 14 /// - Creates a span named "TRANSACTION" with the transaction name 15 /// - Tracks whether the transaction committed or rolled back 16 /// - Automatically rolls back on error (via Drop) 17 /// - Records `otelmart.transaction.outcome` as "commit" or "rollback" 18 /// 19 /// # Arguments 20 /// * `pool` - The database connection pool 21 /// * `name` - A descriptive name for the transaction (e.g., "checkout", "update_order") 22 /// * `f` - The async closure to execute within the transaction 23 /// 24 /// # Example 25 /// ```no_run 26 /// let result = with_transaction(&pool, "checkout", |tx| { 27 /// Box::pin(async move { 28 /// db::create_order(tx, &email, &totals).await?; 29 /// db::create_order_items(tx, order_id, &items).await?; 30 /// Ok(order) 31 /// }) 32 /// }).await?; 33 /// ``` 34 #[instrument( 35 name = "TRANSACTION", 36 skip(pool, f), 37 fields( 38 otel.kind = "client", 39 db.system.name = "postgresql", 40 db.operation.name = "transaction", 41 otelmart.transaction.name = %name, 42 otelmart.transaction.outcome = tracing::field::Empty 43 ) 44 )] 45 pub async fn with_transaction<F, T, E>(pool: &PgPool, name: &'static str, f: F) -> Result<T, E> 46 where 47 F: for<'c> FnOnce( 48 &'c mut Transaction<'_, Postgres>, 49 ) -> Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'c>>, 50 E: From<sqlx::Error>, 51 { 52 let mut tx = pool.begin().await.map_err(E::from)?; 53 54 match f(&mut tx).await { 55 Ok(result) => { 56 tx.commit().await.map_err(E::from)?; 57 Span::current().record("otelmart.transaction.outcome", "commit"); 58 Ok(result) 59 } 60 Err(e) => { 61 // Rollback is automatic on drop, but we make it explicit for clarity 62 // and to ensure the span records the outcome before the error propagates 63 if let Err(rb_err) = tx.rollback().await { 64 tracing::warn!(error = %rb_err, "explicit rollback failed"); 65 } 66 Span::current().record("otelmart.transaction.outcome", "rollback"); 67 Err(e) 68 } 69 } 70 }