Fix a bug in how we handle serialization errors

This commit fixes a bug in how we handle serialization errors with the
postgres backend and transactions. It turns out that we never update the
transaction manager state for `batch_execute` calls, which in turn are
used for executing the transaction SQL itself. That could lead to
situations in which we don't roll back the transaction, but in which we should
have done that.

Fixes #241

Author: weiznich

examples/postgres/pooled-with-rustls/src/main.rs

@@ -41,7 +41,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     Ok(())
 }
 
-fn establish_connection(config: &str) -> BoxFuture<ConnectionResult<AsyncPgConnection>> {
+fn establish_connection(config: &str) -> BoxFuture<'_, ConnectionResult<AsyncPgConnection>> {
     let fut = async {
         // We first set up the way we want rustls to work.
         let rustls_config = ClientConfig::with_platform_verifier();

examples/postgres/run-pending-migrations-with-rustls/src/main.rs

@@ -27,7 +27,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     Ok(())
 }
 
-fn establish_connection(config: &str) -> BoxFuture<ConnectionResult<AsyncPgConnection>> {
+fn establish_connection(config: &str) -> BoxFuture<'_, ConnectionResult<AsyncPgConnection>> {
     let fut = async {
         // We first set up the way we want rustls to work.
         let rustls_config = ClientConfig::with_platform_verifier();

src/mysql/row.rs

@@ -121,7 +121,7 @@ impl<'a> diesel::row::Row<'a, Mysql> for MysqlRow {
         Some(field)
     }
 
-    fn partial_row(&self, range: std::ops::Range<usize>) -> PartialRow<Self::InnerPartialRow> {
+    fn partial_row(&self, range: std::ops::Range<usize>) -> PartialRow<'_, Self::InnerPartialRow> {
         PartialRow::new(self, range)
     }
 }

src/pg/mod.rs

@@ -146,7 +146,12 @@ impl SimpleAsyncConnection for AsyncPgConnection {
             .batch_execute(query)
             .map_err(ErrorHelper)
             .map_err(Into::into);
+
         let r = drive_future(connection_future, batch_execute).await;
+        let r = {
+            let mut transaction_manager = self.transaction_state.lock().await;
+            update_transaction_manager_status(r, &mut transaction_manager)
+        };
         self.record_instrumentation(InstrumentationEvent::finish_query(
             &StrQueryHelper::new(query),
             r.as_ref().err(),
@@ -379,7 +384,7 @@ impl AsyncPgConnection {
     ///     .await
     /// # }
     /// ```
-    pub fn build_transaction(&mut self) -> TransactionBuilder<Self> {
+    pub fn build_transaction(&mut self) -> TransactionBuilder<'_, Self> {
         TransactionBuilder::new(self)
     }
 

src/pg/row.rs

@@ -41,7 +41,7 @@ impl<'a> diesel::row::Row<'a, diesel::pg::Pg> for PgRow {
     fn partial_row(
         &self,
         range: std::ops::Range<usize>,
-    ) -> diesel::row::PartialRow<Self::InnerPartialRow> {
+    ) -> diesel::row::PartialRow<'_, Self::InnerPartialRow> {
         PartialRow::new(self, range)
     }
 }

src/transaction_manager.rs

@@ -381,12 +381,8 @@ where
                     ..
                 }) = conn.transaction_state().status
                 {
-                    match Self::critical_transaction_block(
-                        &is_broken,
-                        Self::rollback_transaction(conn),
-                    )
-                    .await
-                    {
+                    // rollback_transaction handles the critical block internally on its own
+                    match Self::rollback_transaction(conn).await {
                         Ok(()) => {}
                         Err(rollback_error) => {
                             conn.transaction_state().status.set_in_error();

tests/lib.rs

@@ -11,6 +11,7 @@ mod instrumentation;
 mod pooling;
 #[cfg(feature = "async-connection-wrapper")]
 mod sync_wrapper;
+mod transactions;
 mod type_check;
 
 async fn transaction_test<C: AsyncConnection<Backend = TestBackend>>(

tests/transactions.rs

@@ -0,0 +1,106 @@
+#[cfg(feature = "postgres")]
+#[tokio::test]
+async fn concurrent_serializable_transactions_behave_correctly() {
+    use diesel::prelude::*;
+    use diesel_async::RunQueryDsl;
+    use std::sync::Arc;
+    use tokio::sync::Barrier;
+
+    table! {
+        users3 {
+            id -> Integer,
+        }
+    }
+
+    // create an async connection
+    let mut conn = super::connection_without_transaction().await;
+
+    let mut conn1 = super::connection_without_transaction().await;
+
+    diesel::sql_query("CREATE TABLE IF NOT EXISTS users3 (id int);")
+        .execute(&mut conn)
+        .await
+        .unwrap();
+
+    let barrier_1 = Arc::new(Barrier::new(2));
+    let barrier_2 = Arc::new(Barrier::new(2));
+    let barrier_1_for_tx1 = barrier_1.clone();
+    let barrier_1_for_tx2 = barrier_1.clone();
+    let barrier_2_for_tx1 = barrier_2.clone();
+    let barrier_2_for_tx2 = barrier_2.clone();
+
+    let mut tx = conn.build_transaction().serializable().read_write();
+
+    let res = tx.run(|conn| {
+        Box::pin(async {
+            users3::table.select(users3::id).load::<i32>(conn).await?;
+
+            barrier_1_for_tx1.wait().await;
+            diesel::insert_into(users3::table)
+                .values(users3::id.eq(1))
+                .execute(conn)
+                .await?;
+            barrier_2_for_tx1.wait().await;
+
+            Ok::<_, diesel::result::Error>(())
+        })
+    });
+
+    let mut tx1 = conn1.build_transaction().serializable().read_write();
+
+    let res1 = async {
+        let res = tx1
+            .run(|conn| {
+                Box::pin(async {
+                    users3::table.select(users3::id).load::<i32>(conn).await?;
+
+                    barrier_1_for_tx2.wait().await;
+                    diesel::insert_into(users3::table)
+                        .values(users3::id.eq(1))
+                        .execute(conn)
+                        .await?;
+
+                    Ok::<_, diesel::result::Error>(())
+                })
+            })
+            .await;
+        barrier_2_for_tx2.wait().await;
+        res
+    };
+
+    let (res, res1) = tokio::join!(res, res1);
+    let _ = diesel::sql_query("DROP TABLE users3")
+        .execute(&mut conn1)
+        .await;
+
+    assert!(
+        res1.is_ok(),
+        "Expected the second transaction to be succussfull, but got an error: {:?}",
+        res1.unwrap_err()
+    );
+
+    assert!(res.is_err(), "Expected the first transaction to fail");
+    let err = res.unwrap_err();
+    assert!(
+        matches!(
+            &err,
+            diesel::result::Error::DatabaseError(
+                diesel::result::DatabaseErrorKind::SerializationFailure,
+                _
+            )
+        ),
+        "Expected an serialization failure but got another error: {err:?}"
+    );
+
+    let mut tx = conn.build_transaction();
+
+    let res = tx
+        .run(|_| Box::pin(async { Ok::<_, diesel::result::Error>(()) }))
+        .await;
+
+    assert!(
+        res.is_ok(),
+        "Expect transaction to run fine but got an error: {:?}",
+        res.unwrap_err()
+    );
+}