Connection::transaction is not usable

[weiznich]

The following code fails to compile:

async fn test_transaction(conn: &mut SqliteConnection) {
    use sqlx::Connection;
    conn.transaction(|conn| {
        async {
            sqlx::query("SELECT * FROM foo;")
                .execute(&mut *conn)
                .await
                .unwrap();
            sqlx::query("SELECT * FROM foo;")
                .execute(&mut *conn)
                .await
                .unwrap();
            sqlx::Result::Ok(())
        }
    })
    .await
    .unwrap();
}

with the corresponding error message:

error: lifetime may not live long enough
   --> example.rs:158:9
    |
157 |       conn.transaction(|conn| {
    |                         ----- return type of closure is impl std::future::Future
    |                         |
    |                         has type `&'1 mut sqlx_core::sqlite::connection::SqliteConnection`
158 | /         async {
159 | |             sqlx::query("SELECT * FROM foo;")
160 | |                 .execute(&mut *conn)
161 | |                 .await
...   |
167 | |             sqlx::Result::Ok(())
168 | |         }
    | |_________^ returning this value requires that `'1` must outlive `'2`

error[E0373]: async block may outlive the current function, but it borrows `conn`, which is owned by the current function
   --> example.rs:158:15
    |
158 |           async {
    |  _______________^
159 | |             sqlx::query("SELECT * FROM foo;")
160 | |                 .execute(&mut *conn)
    | |                                ---- `conn` is borrowed here
161 | |                 .await
...   |
167 | |             sqlx::Result::Ok(())
168 | |         }
    | |_________^ may outlive borrowed value `conn`
    |

Environment:

• sqlx: 0.4.0-beta.1
• rustc 1.45.0 (5c1f21c3b 2020-07-13)
• OS: Ubuntu 20.04
• Backend: Sqlite (does not really matter)

[mehcode]

You need:

conn.transaction(|conn| async move {
  // [...]
})

Otherwise the inner async block borrows the connection reference from the closure and then returns the borrow which would be borrowing a temporary.

[weiznich]

That does not change much:

error: lifetime may not live long enough
   --> example.rs:305:9
    |
304 |       conn.transaction(|conn| {
    |                         ----- return type of closure is impl std::future::Future
    |                         |
    |                         has type `&'1 mut sqlx_core::sqlite::connection::SqliteConnection`
305 | /         async move {
306 | |             sqlx::query("SELECT * FROM foo;")
307 | |                 .execute(&mut *conn)
308 | |                 .await
...   |
314 | |             sqlx::Result::Ok(())
315 | |         }
    | |_________^ returning this value requires that `'1` must outlive `'2`

The underlying issue is that the lifetime bounds on Connection::transaction are "wrong". Instead of
F: FnOnce(&mut <Self::Database as Database>::Connection) -> Fut + Send + 'f you would need to write something like
F: for<'a> FnOnce(&'a mut <Self::Database as Database>::Connection) -> impl Future<Output = Result<T, E> + 'a> to signal that what is returned by your future cannot reference anything which is passed into the closure. Unfortunately such a bound is currently not supported by rustc, so this API would need the language to be changed to make it working.

As this method was introduced with 0.4.0-beta.1 and it is in the current form not usable maybe it should be removed before the 0.4.0 release?

[Kixiron]

This makes complex Sqlite databases impossible to use, since foreign key constraints throw errors on failure and can only be deferred within the context of a transaction

[jyn514]

@Kixiron you can use begin instead, you just can't wrap the whole thing in one closure: https://docs.rs/sqlx/0.4.0-beta.1/sqlx/trait.Connection.html#tymethod.begin

[argv-minus-one]

Rust currently doesn't support this. Generic associated types are needed to make it work, apparently.

The current workaround for this problem is to box the future, as I've done in my test code:

pub async fn database_test(
	f: impl for<'tx> FnOnce(&'tx mut sqlx::Transaction<<Connection as sqlx::Connection>::Database>) -> Pin<Box<dyn Future<Output = ()> + 'tx>>,
) {
	…
}

#[tokio::test]
async fn some_test() {
	database_test(|dbc| Box::pin(async move {
		…
	})).await;
}

[OskarPersson]

@argv-minus-one Could you share your database_test implementation? I'm looking to do something similar :)

[mehcode]

This is fixed by #878

[argv-minus-one]

Here's my current implementation (with some unimportant application-specific bits stripped out).

pub async fn database_test(
	f: impl for<'tx> FnOnce(&'tx mut sqlx::Transaction<sqlx::Postgres>) -> Pin<Box<dyn Future<Output = ()> + 'tx>>,
) {
	// Create a test database.
	temp_db(
		&todo!(), // substitute your own `PgConnectOptions` here
		|_, dbc| async move {
			// Start the test transaction.
			let mut tx = dbc.begin().await
				.expect("couldn't start test transaction");

			// Run the test function.
			f(&mut tx).await;

			// Roll back the test transaction.
			tx.rollback().await
				.expect("couldn't rollback test transaction");
		},
	).await.unwrap()
}

This uses another piece of code to create a temporary database in which to run a test, then drop it once the test is done. That way, every test is run in a clean, consistent database environment. Here it is:

use futures::FutureExt as _;
use rand::Rng as _;
use sqlx::{
	Connection as _,
	Executor as _,
	postgres::{
		PgConnection,
		PgConnectOptions,
		PgPool,
	},
};
use std::{
	future::Future,
	panic::{AssertUnwindSafe, resume_unwind},
};

#[derive(Debug, thiserror::Error)]
pub enum TempDbError {
	#[error("couldn't create temporary database: {0}")]
	Create(#[source] sqlx::Error),

	#[error("couldn't connect to temporary database “{name}”: {error}")]
	Connect {
		#[source] error: sqlx::Error,
		name: String,
	},

	#[error("couldn't drop temporary database “{name}”: {error}")]
	Drop {
		#[source] error: sqlx::Error,
		name: String,
	},
}

/// Runs an async closure with a temporary database.
///
/// This function does the following:
///
/// 1. Create a temporary database using the provided `connect_options`.
/// 2. Open a pool of connections to the temporary database.
/// 3. Call `f` with the connection pool.
/// 4. Await the future returned by `f`, catching panic.
/// 5. Close all of the connections in the pool.
/// 6. Drop the temporary database, again using the provided `connect_options`.
/// 7. If `f` panicked, resume unwinding. If not, return the result.
///
/// # Errors
///
/// If creating, connecting to, or dropping the temporary database fails, the resulting error will be returned. If creating or connecting to the temporary database fails, `f` will not be called.
///
/// If connecting to the temporary database fails, it will be assumed that dropping the temporary database is also impossible. In this case, the temporary database will be left behind without being dropped.
///
/// If creating or dropping the temporary database succeeds, but then there's an error closing the database connection that was used to create or drop it, then this function will print a message to stderr but continue on.
pub async fn temp_db<T, Fut>(
	connect_options: &PgConnectOptions,
	f: impl FnOnce(String, PgPool) -> Fut,
) -> Result<T, TempDbError>
where
	Fut: Future<Output = T>,
{
	let mut dbc = PgConnection::connect_with(connect_options).await.map_err(TempDbError::Create)?;

	let temp_db_name: sqlx::Result<String> = async {
		loop {
			fn temp_db_name() -> String {
				format!(
					"temp{}",
					rand::thread_rng().gen::<u16>(),
				)
			}

			let temp_db_name = temp_db_name();

			if let Err(error) = dbc.execute(format!("CREATE DATABASE \"{}\"", temp_db_name).as_str()).await {
				let is_dup_db: bool =
					error.as_database_error()
					.and_then(|error| error.code())
					.map(|code| code.as_ref() == "42P04")
					.unwrap_or(false);

				if !is_dup_db {
					return Err(error);
				}
			}
			else {
				break Ok(temp_db_name);
			}
		}
	}.await;

	// Close the connection, whether or not the test database was created successfully.
	if let Err(error) = dbc.close().await {
		// There are two ways we can end up here:
		//
		// 1. Creating the test database failed, and closing the connection also failed.
		// 2. Creating the test database succeeded, but closing the connection failed.
		//
		// In either case, we should not bail just because closing the connection fails, because:
		//
		// 1. If the test database was created successfully, then we still need to try to drop it. That involves opening another database connection later anyway, so we may as well carry on.
		// 2. If the test database was not created successfully, then we'll bail with that error below.
		//
		// We may want to print a warning now, depending on which scenario happened:
		//
		// 1. If the test database was created successfully, then we'll carry on, so we should print a warning.
		// 2. If the test database was not created successfully, then we're going to bail anyway, so there's no point in warning about the connection.

		if temp_db_name.is_ok() {
			eprintln!("Error closing database connection after creating temporary database: {}", error);
		}
	}

	// Unwrap the temporary database name.
	let temp_db_name: String = temp_db_name.map_err(TempDbError::Create)?;

	// Open a connection pool to the temporary database.
	let pool = match PgPool::connect_with(connect_options.clone().database(temp_db_name.as_str())).await {
		Ok(ok) => ok,
		Err(error) => return Err(TempDbError::Connect {
			error,
			name: temp_db_name,
		}),
	};

	// Run the provided function. Assert that it's unwind-safe, because we'll resume unwinding before anyone gets a chance to observe any invalid state that `f` leaves behind.
	let result = AssertUnwindSafe(f(temp_db_name.clone(), pool.clone())).catch_unwind().await;

	// Close the connection pool. This has to happen before dropping the temporary database, because PostgreSQL will refuse to drop the temporary database if there are any live connections left.
	pool.close().await;

	// Drop the temporary database.
	let drop_result: Result<(), TempDbError> = async {
		// Open a connection.
		let mut dbc = PgConnection::connect_with(connect_options).await?;

		// Send the drop command, and get the result.
		let drop_result = dbc.execute(format!("DROP DATABASE \"{}\"", temp_db_name).as_str()).await.map(|_| ());

		// Close the connection. Do this even if the drop command fails.
		let close_result = dbc.close().await;

		// Return the first error that occurred.
		drop_result.or(close_result)
	}.await.map_err(|error| TempDbError::Drop {
		error,
		name: temp_db_name,
	});

	// If the provided function panicked, resume unwinding. Otherwise, return the result of dropping the temporary database.
	match result {
		Ok(ok) => {
			drop_result?;
			Ok(ok)
		}

		Err(panic) => resume_unwind(panic)
	}
}

Edit: Completely different temporary database routine. The previous one would leave behind temporary databases if there was a panic during test execution, because in that case, PostgreSQL would refuse to drop the temporary database as there was still a live connection to it! This one seems to work correctly even if there is a panic, but who knows if I'll find some other problem with it. Database testing is hard…

[mehcode]

This should be fixed in master.