Merge pull request #242 from weiznich/fix/241

Fix a bug in how we handle serialization errors

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()
+    );
+}