feat(postgres): add sqlx-advisory-locking connection string parameter honored in #[sqlx::test]

sqlx-core/src/testing/mod.rs

@@ -82,6 +82,7 @@ pub struct TestContext<DB: Database> {
     pub pool_opts: PoolOptions<DB>,
     pub connect_opts: <DB::Connection as Connection>::Options,
     pub db_name: String,
+    pub locking: bool,
 }
 
 impl<DB, Fut> TestFn for fn(Pool<DB>) -> Fut
@@ -226,7 +227,7 @@ where
             .await
             .expect("failed to connect to setup test database");
 
-        setup_test_db::<DB>(&test_context.connect_opts, &args).await;
+        setup_test_db::<DB>(&test_context.connect_opts, &args, test_context.locking).await;
 
         let res = test_fn(test_context.pool_opts, test_context.connect_opts).await;
 
@@ -246,6 +247,7 @@ where
 async fn setup_test_db<DB: Database>(
     copts: &<DB::Connection as Connection>::Options,
     args: &TestArgs,
+    locking: bool,
 ) where
     DB::Connection: Migrate + Sized,
     for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>,
@@ -255,7 +257,24 @@ async fn setup_test_db<DB: Database>(
         .await
         .expect("failed to connect to test database");
 
-    if let Some(migrator) = args.migrator {
+    if let Some(static_migrator) = args.migrator {
+        // When advisory locking is disabled, also disable it in the migrator.
+        // This is required for databases that don't support advisory locks (e.g. CockroachDB).
+        let owned_migrator;
+        let migrator = if locking {
+            static_migrator
+        } else {
+            owned_migrator = Migrator {
+                locking: false,
+                migrations: static_migrator.migrations.clone(),
+                ignore_missing: static_migrator.ignore_missing,
+                no_tx: static_migrator.no_tx,
+                table_name: static_migrator.table_name.clone(),
+                create_schemas: static_migrator.create_schemas.clone(),
+            };
+            &owned_migrator
+        };
+
         migrator
             .run_direct(None, &mut conn)
             .await

sqlx-mysql/src/testing/mod.rs

@@ -178,6 +178,7 @@ async fn test_context(args: &TestArgs) -> Result<TestContext<MySql>, Error> {
             .clone()
             .database(&db_name),
         db_name,
+        locking: true,
     })
 }
 

sqlx-postgres/src/options/doc.md

@@ -48,6 +48,7 @@ Instead, the name is linked to the method to set the value.
 | Parameter                                                    | Default                       |
 |--------------------------------------------------------------|-------------------------------|
 | [`statement-cache-capacity`][Self::statement_cache_capacity] | `100`                         |
+| [`sqlx-advisory-locking`][Self::advisory_locking]            | `true`                        |
 
 # Example URLs
 ```text

sqlx-postgres/src/options/mod.rs

@@ -30,6 +30,7 @@ pub struct PgConnectOptions {
     pub(crate) log_settings: LogSettings,
     pub(crate) extra_float_digits: Option<Cow<'static, str>>,
     pub(crate) options: Option<String>,
+    pub(crate) advisory_locking: bool,
 }
 
 impl Default for PgConnectOptions {
@@ -97,6 +98,7 @@ impl PgConnectOptions {
             extra_float_digits: Some("2".into()),
             log_settings: Default::default(),
             options: var("PGOPTIONS").ok(),
+            advisory_locking: true,
         }
     }
 
@@ -452,6 +454,19 @@ impl PgConnectOptions {
         self
     }
 
+    /// Sets whether `#[sqlx::test]` should use advisory locking during test database setup.
+    ///
+    /// Defaults to `true`. Set to `false` for databases that speak the PostgreSQL wire
+    /// protocol but do not implement `pg_advisory_xact_lock`, such as CockroachDB.
+    ///
+    /// When disabled, advisory locking is also skipped for the migrator during test setup.
+    ///
+    /// Can be set in the connection URL: `postgres://localhost/mydb?sqlx-advisory-locking=false`
+    pub fn advisory_locking(mut self, advisory_locking: bool) -> Self {
+        self.advisory_locking = advisory_locking;
+        self
+    }
+
     /// We try using a socket if hostname starts with `/` or if socket parameter
     /// is specified.
     pub(crate) fn fetch_socket(&self) -> Option<String> {
@@ -580,6 +595,11 @@ impl PgConnectOptions {
     pub fn get_options(&self) -> Option<&str> {
         self.options.as_deref()
     }
+
+    /// Get whether advisory locking is enabled for test database setup.
+    pub fn get_advisory_locking(&self) -> bool {
+        self.advisory_locking
+    }
 }
 
 fn default_host(port: u16) -> String {

sqlx-postgres/src/options/parse.rs

@@ -104,6 +104,10 @@ impl PgConnectOptions {
                     }
                 }
 
+                "sqlx-advisory-locking" => {
+                    options = options.advisory_locking(value.parse().map_err(Error::config)?);
+                }
+
                 _ => tracing::warn!(%key, %value, "ignoring unrecognized connect parameter"),
             }
         }
@@ -340,3 +344,25 @@ fn built_url_can_be_parsed() {
 
     assert!(parsed.is_ok());
 }
+
+#[test]
+fn it_parses_advisory_locking_correctly() {
+    let url = "postgres://localhost?sqlx-advisory-locking=false";
+    let opts = PgConnectOptions::from_str(url).unwrap();
+    assert!(!opts.advisory_locking);
+
+    let url = "postgres://localhost?sqlx-advisory-locking=true";
+    let opts = PgConnectOptions::from_str(url).unwrap();
+    assert!(opts.advisory_locking);
+
+    // Works after other parameters
+    let url = "postgres://localhost?application_name=myapp&sqlx-advisory-locking=false";
+    let opts = PgConnectOptions::from_str(url).unwrap();
+    assert!(!opts.advisory_locking);
+    assert_eq!(opts.application_name.as_deref(), Some("myapp"));
+
+    // Default is true
+    let url = "postgres://localhost";
+    let opts = PgConnectOptions::from_str(url).unwrap();
+    assert!(opts.advisory_locking);
+}

sqlx-postgres/src/testing/mod.rs

@@ -93,6 +93,7 @@ async fn test_context(args: &TestArgs) -> Result<TestContext<Postgres>, Error> {
     let url = dotenvy::var("DATABASE_URL").expect("DATABASE_URL must be set");
 
     let master_opts = PgConnectOptions::from_str(&url).expect("failed to parse DATABASE_URL");
+    let advisory_locking = master_opts.advisory_locking;
 
     let pool = PoolOptions::new()
         // Postgres' normal connection limit is 100 plus 3 superuser connections
@@ -125,31 +126,44 @@ async fn test_context(args: &TestArgs) -> Result<TestContext<Postgres>, Error> {
 
     let mut conn = master_pool.acquire().await?;
 
-    // language=PostgreSQL
-    conn.execute(
-        // Explicit lock avoids this latent bug: https://stackoverflow.com/a/29908840
-        // I couldn't find a bug on the mailing list for `CREATE SCHEMA` specifically,
-        // but a clearly related bug with `CREATE TABLE` has been known since 2007:
-        // https://www.postgresql.org/message-id/200710222037.l9MKbCJZ098744%40wwwmaster.postgresql.org
-        // magic constant 8318549251334697844 is just 8 ascii bytes 'sqlxtest'.
+    // Explicit lock avoids this latent bug: https://stackoverflow.com/a/29908840
+    // I couldn't find a bug on the mailing list for `CREATE SCHEMA` specifically,
+    // but a clearly related bug with `CREATE TABLE` has been known since 2007:
+    // https://www.postgresql.org/message-id/200710222037.l9MKbCJZ098744%40wwwmaster.postgresql.org
+    // magic constant 8318549251334697844 is just 8 ascii bytes 'sqlxtest'.
+    //
+    // The lock and DDL must be in a single `execute` call so they share one implicit
+    // transaction — `pg_advisory_xact_lock` is released at transaction end.
+    //
+    // Can be disabled with `?sqlx-advisory-locking=false` in DATABASE_URL for databases
+    // that do not implement advisory locks, such as CockroachDB.
+    let lock_sql = if advisory_locking {
+        "select pg_advisory_xact_lock(8318549251334697844);"
+    } else {
+        ""
+    };
+
+    let setup_sql = format!(
+        // language=PostgreSQL
         r#"
-        select pg_advisory_xact_lock(8318549251334697844);
+            {lock_sql}
+
+            create schema if not exists _sqlx_test;
 
-        create schema if not exists _sqlx_test;
+            create table if not exists _sqlx_test.databases (
+                db_name text primary key,
+                test_path text not null,
+                created_at timestamptz not null default now()
+            );
 
-        create table if not exists _sqlx_test.databases (
-            db_name text primary key,
-            test_path text not null,
-            created_at timestamptz not null default now()
-        );
+            create index if not exists databases_created_at
+                on _sqlx_test.databases(created_at);
 
-        create index if not exists databases_created_at 
-            on _sqlx_test.databases(created_at);
+            create sequence if not exists _sqlx_test.database_ids;
+        "#
+    );
 
-        create sequence if not exists _sqlx_test.database_ids;
-    "#,
-    )
-    .await?;
+    conn.execute(AssertSqlSafe(setup_sql)).await?;
 
     let db_name = Postgres::db_name(args);
     do_cleanup(&mut conn, &db_name).await?;
@@ -183,6 +197,7 @@ async fn test_context(args: &TestArgs) -> Result<TestContext<Postgres>, Error> {
             .clone()
             .database(&db_name),
         db_name,
+        locking: advisory_locking,
     })
 }
 

sqlx-sqlite/src/testing/mod.rs

@@ -58,6 +58,7 @@ async fn test_context(args: &TestArgs) -> Result<TestContext<Sqlite>, Error> {
         // The main limitation is going to be the number of concurrent running tests.
         pool_opts: PoolOptions::new().max_connections(1000),
         db_name: db_path,
+        locking: true,
     })
 }
 

src/macros/test.md

@@ -226,4 +226,28 @@ Multiple `fixtures` attributes can be used to combine different operating modes.
 <sup>3</sup>Ordering for test fixtures is entirely up to the application, and each test may choose which fixtures to
 apply and which to omit. However, since each fixture is applied separately (sent as a single command string, so wrapped
 in an implicit `BEGIN` and `COMMIT`), you will want to make sure to order the fixtures such that foreign key
-requirements are always satisfied, or else you might get errors. 
+requirements are always satisfied, or else you might get errors.
+
+### Disabling Advisory Locking
+
+By default, `#[sqlx::test]` acquires a PostgreSQL advisory lock (`pg_advisory_xact_lock`) before creating the test
+database schema and also locks during migrations. This prevents a race condition when multiple tests run concurrently.
+
+Some databases speak the PostgreSQL wire protocol but do not implement advisory locks.
+For example, CockroachDB does not support `pg_advisory_xact_lock`
+(see [cockroachdb/cockroach#13546](https://github.com/cockroachdb/cockroach/issues/13546)).
+
+You can disable advisory locking by adding `sqlx-advisory-locking=false` to your `DATABASE_URL`:
+
+```text
+DATABASE_URL="postgres://localhost/mydb?sqlx-advisory-locking=false"
+```
+
+This affects all `#[sqlx::test]` tests: both the test database schema setup and migrations
+will skip advisory locks. No per-test annotation is needed.
+
+See also [`PgConnectOptions::advisory_locking()`](crate::postgres::PgConnectOptions::advisory_locking)
+and [`Migrator::set_locking()`](crate::migrate::Migrator::set_locking).
+
+**Note:** Disabling locking means concurrent test processes may race during schema setup. The DDL statements
+use `IF NOT EXISTS` so this is generally safe, but you should be aware of the trade-off.

tests/postgres/test-attr.rs

@@ -199,3 +199,25 @@ macro_rules! macro_using_test {
     };
 }
 macro_using_test!("tests/postgres/migrations");
+
+// Verifies that `?sqlx-advisory-locking=false` in DATABASE_URL is respected.
+//
+// This test is #[ignore] because it cannot safely run concurrently with locked tests on
+// the same Postgres server. The advisory lock guards a DDL race that only manifests under
+// concurrency on standard PostgreSQL. The feature is intended for databases like CockroachDB
+// that serialize DDL and don't support advisory locks.
+//
+// To run: DATABASE_URL="postgres://...?sqlx-advisory-locking=false" cargo test -- --ignored it_works_without_locking
+#[sqlx::test(migrations = false)]
+#[ignore]
+async fn it_works_without_locking(pool: PgPool) -> sqlx::Result<()> {
+    let mut conn = pool.acquire().await?;
+
+    let db_name: String = sqlx::query_scalar("SELECT current_database()")
+        .fetch_one(&mut *conn)
+        .await?;
+
+    assert!(db_name.starts_with("_sqlx_test"), "dbname: {db_name:?}");
+
+    Ok(())
+}