fix(core): accept transaction callbacks without Sync

sqlx-core/src/connection.rs

@@ -103,8 +103,7 @@ pub trait Connection: Send {
     where
         for<'c> F: FnOnce(&'c mut Transaction<'_, Self::Database>) -> BoxFuture<'c, Result<R, E>>
             + 'a
-            + Send
-            + Sync,
+            + Send,
         Self: Sized,
         R: Send,
         E: From<Error> + Send,

sqlx-test/src/lib.rs

@@ -1,5 +1,5 @@
 use sqlx::pool::PoolOptions;
-use sqlx::{Connection, Database, Error, Pool};
+use sqlx::{Connection, Database, Error, Executor, Pool};
 use std::env;
 
 pub fn setup_if_needed() {
@@ -38,6 +38,59 @@ where
     Ok(pool)
 }
 
+/// Check that Send-only transaction callbacks can commit and roll back.
+pub async fn test_send_only_transaction_callbacks<C>(conn: &mut C) -> anyhow::Result<()>
+where
+    C: Connection,
+    for<'c> &'c mut C: Executor<'c, Database = C::Database>,
+{
+    fn assert_send(_: &(impl std::future::Future + Send)) {}
+
+    sqlx::raw_sql("CREATE TEMPORARY TABLE transaction_test (value BIGINT)")
+        .execute(&mut *conn)
+        .await?;
+
+    for commit in [true, false] {
+        // A Receiver is Send but not Sync.
+        let (sender, receiver) = std::sync::mpsc::channel();
+        sender.send(42_i64)?;
+
+        let future = conn.transaction(move |tx| {
+            Box::pin(async move {
+                sqlx::raw_sql("INSERT INTO transaction_test (value) VALUES (42)")
+                    .execute(&mut **tx)
+                    .await?;
+
+                // Keep the receiver across an await in the callback future, too.
+                let value = receiver.try_recv().unwrap();
+
+                if commit {
+                    Ok(value)
+                } else {
+                    Err(sqlx::Error::RowNotFound)
+                }
+            })
+        });
+
+        assert_send(&future);
+        let result = future.await;
+
+        if commit {
+            assert_eq!(result?, 42);
+        } else {
+            assert!(matches!(result, Err(sqlx::Error::RowNotFound)));
+        }
+
+        assert!(!conn.is_in_transaction());
+        let rows = sqlx::raw_sql("SELECT value FROM transaction_test")
+            .fetch_all(&mut *conn)
+            .await?;
+        assert_eq!(rows.len(), 1);
+    }
+
+    Ok(())
+}
+
 // Test type encoding and decoding
 #[macro_export]
 macro_rules! test_type {

tests/any/any.rs

@@ -23,6 +23,14 @@ async fn it_connects() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[sqlx_macros::test]
+async fn it_accepts_send_only_transaction_callbacks() -> anyhow::Result<()> {
+    sqlx::any::install_default_drivers();
+    let mut conn = new::<Any>().await?;
+
+    sqlx_test::test_send_only_transaction_callbacks(&mut conn).await
+}
+
 #[sqlx_macros::test]
 async fn it_pings() -> anyhow::Result<()> {
     sqlx::any::install_default_drivers();

tests/mysql/mysql.rs

@@ -19,6 +19,13 @@ async fn it_connects() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[sqlx_macros::test]
+async fn it_accepts_send_only_transaction_callbacks() -> anyhow::Result<()> {
+    let mut conn = new::<MySql>().await?;
+
+    sqlx_test::test_send_only_transaction_callbacks(&mut conn).await
+}
+
 #[sqlx_macros::test]
 async fn it_connects_without_password() -> anyhow::Result<()> {
     setup_if_needed();

tests/postgres/postgres.rs

@@ -28,6 +28,13 @@ async fn it_connects() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[sqlx_macros::test]
+async fn it_accepts_send_only_transaction_callbacks() -> anyhow::Result<()> {
+    let mut conn = new::<Postgres>().await?;
+
+    sqlx_test::test_send_only_transaction_callbacks(&mut conn).await
+}
+
 #[sqlx_macros::test]
 async fn it_can_select_void() -> anyhow::Result<()> {
     let mut conn = new::<Postgres>().await?;

tests/sqlite/sqlite.rs

@@ -1344,6 +1344,13 @@ async fn test_serialize_invalid_schema() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[sqlx_macros::test]
+async fn it_accepts_send_only_transaction_callbacks() -> anyhow::Result<()> {
+    let mut conn = SqliteConnection::connect(":memory:").await?;
+
+    sqlx_test::test_send_only_transaction_callbacks(&mut conn).await
+}
+
 #[sqlx_macros::test]
 async fn it_can_use_transaction_options() -> anyhow::Result<()> {
     async fn check_txn_state(conn: &mut SqliteConnection, expected: SqliteTransactionState) {