fix(tests): install eql_v3 via sqlx Migrator, not a hand-rolled installer (CIP-3141)

c012fdc made the property suites self-install eql_v3 with an embedded copy of
the release SQL guarded by a hand-rolled advisory lock. That raced: nextest
runs each test in its OWN process, several hit the empty shard DB at once, and
concurrent `CREATE SCHEMA eql_v3` failed with `duplicate key … pg_namespace`.
The lock attempt was reinventing — incorrectly — what sqlx already does.

Use the real thing: `ensure_eql_installed` now runs `sqlx::migrate!("./migrations")`
against the base pool — the SAME embedded migration set `#[sqlx::test]` applies
to every other test in the suite. `Migrator::run` records applied versions in
`_sqlx_migrations` and skips them (idempotent, so a developer's pre-migrated
local DB is a no-op) and holds a database-level advisory lock for the run, so
the per-test processes serialise: exactly one applies each migration, the rest
observe it already applied. No bespoke installer, no bespoke lock.

The migrator is built in the test target (property/mod.rs `migrator()`) so the
lib never embeds the gitignored generated `001_install_eql.sql`. Keeps the
`{e:#}` change from c012fdc that surfaces anyhow's cause chain (it is what
revealed the duplicate-key error in the first place).

Compiles clean; CI's sharded nextest run (process-per-test) is the verification
that the migrator serialises the concurrent install.

Author: tobyhede

tests/sqlx/src/property.rs

@@ -22,49 +22,30 @@ use tokio::sync::Mutex;
 /// `fixtures.eql_v2_<T>` table exactly once.
 static FIXTURE_LOADED: OnceLock<Mutex<HashSet<&'static str>>> = OnceLock::new();
 
-/// Per-process guard ensuring the EQL surface (the `eql_v3` schema the oracle
-/// queries cast to) is installed into the connected DB exactly once.
-static EQL_INSTALLED: OnceLock<Mutex<bool>> = OnceLock::new();
-
-/// Ensure the EQL surface (the `eql_v3` schema + scalar domains/operators the
-/// oracle queries cast to) is present in the DB behind `pool`.
+/// Apply the SQLx migrations (the EQL install in `001_install_eql.sql`, plus the
+/// regression-data migrations) to the DB behind `pool`.
 ///
 /// The property suites connect via `connect_pool()` to the base test database
-/// (`DATABASE_URL`), NOT through `#[sqlx::test]`'s migrated per-test scratch
-/// DBs. In a CI shard that base DB is a stock Postgres with no EQL installed —
-/// only the `build-archive` job ran `sqlx migrate run`, and against a different
-/// Postgres — so every `::eql_v3.<T>_eq` cast would raise
-/// `schema "eql_v3" does not exist`. This installs the surface on demand so the
-/// suites are self-sufficient regardless of where they run (CI shard, local,
-/// fork), instead of silently depending on a pre-installed base DB.
+/// (`DATABASE_URL`) rather than through `#[sqlx::test]`'s migrated per-test
+/// scratch DBs, because their proptest case loop is synchronous and cannot take
+/// `#[sqlx::test]`'s injected pool. In a CI shard that base DB is a stock
+/// Postgres with no EQL installed, so every `::eql_v3.<T>_eq` cast would raise
+/// `schema "eql_v3" does not exist`. This brings the base DB up to the same
+/// migrated state the rest of the suite gets for free.
 ///
-/// `install_sql` is the EQL installer (`migrations/001_install_eql.sql`),
-/// `include_str!`-embedded into the test binary at compile time (see
-/// `property/mod.rs`) so it travels inside the prebuilt nextest archive — the
-/// same mechanism the fixture corpus uses. A process-wide async mutex
-/// guarantees exactly-once execution across the parallel proptest threads, and
-/// a presence check (`eql_v3.int4_eq`) skips the install when the DB already
-/// has the surface (a developer's pre-installed local DB), where re-running the
-/// non-idempotent installer would error on duplicate objects.
-pub async fn ensure_eql_installed(pool: &PgPool, install_sql: &str) -> Result<()> {
-    let guard = EQL_INSTALLED.get_or_init(|| Mutex::new(false));
-    let mut installed = guard.lock().await;
-    if *installed {
-        return Ok(());
-    }
-    // Presence check: skip the installer if the surface is already there. int4
-    // is the reference scalar type and is always part of the surface.
-    let present: bool = sqlx::query_scalar("SELECT to_regtype('eql_v3.int4_eq') IS NOT NULL")
-        .fetch_one(pool)
+/// `migrator` is `sqlx::migrate!("./migrations")` — the SAME embedded migration
+/// set `#[sqlx::test]` runs, passed in from the test binary so the lib does not
+/// embed the (gitignored, generated) migration files. `Migrator::run` is
+/// idempotent (it records applied versions in `_sqlx_migrations` and skips
+/// them) and process-safe (it takes a database-level advisory lock for the
+/// duration), so the separate OS processes nextest runs each test in serialise
+/// correctly — exactly one applies each migration, the rest observe it already
+/// applied. A developer's already-migrated local DB is a no-op.
+pub async fn ensure_eql_installed(pool: &PgPool, migrator: &sqlx::migrate::Migrator) -> Result<()> {
+    migrator
+        .run(pool)
         .await
-        .context("probing for an existing eql_v3 install")?;
-    if !present {
-        sqlx::raw_sql(install_sql)
-            .execute(pool)
-            .await
-            .context("installing the EQL surface into the property-test DB")?;
-    }
-    *installed = true;
+        .context("applying EQL migrations to the property-test DB")?;
     Ok(())
 }
 

tests/sqlx/tests/encrypted_domain/property/e2e_oracle.rs

@@ -67,8 +67,9 @@ where
         .build()?;
     let pool: PgPool = rt.block_on(connect_pool())?;
     // The base DB this pool connects to is not migrated by `#[sqlx::test]`; in a
-    // CI shard it has no `eql_v3` surface, so install it before any cast/query.
-    rt.block_on(ensure_eql_installed(&pool, super::EQL_INSTALL_SQL))?;
+    // CI shard it has no `eql_v3` surface, so apply the migrations (idempotent +
+    // process-safe via the migrator's advisory lock) before any cast/query.
+    rt.block_on(ensure_eql_installed(&pool, &super::migrator()))?;
 
     // Shrinking is disabled for the e2e suite: every failed shrink attempt would
     // trigger another ZeroKMS batch, and ciphertext cannot be meaningfully

tests/sqlx/tests/encrypted_domain/property/fixture_oracle.rs

@@ -65,8 +65,9 @@ fn embedded_fixture_sql<T: ScalarType>() -> &'static str {
 /// `#[sqlx::test]`'s ephemeral DBs by default, not the pool we connect to).
 async fn load_fixture_rows<T: ScalarType>(pool: &PgPool) -> Result<Vec<Row<T>>> {
     // The base DB this pool connects to is not migrated by `#[sqlx::test]`; in a
-    // CI shard it has no `eql_v3` surface, so install it before any cast/query.
-    ensure_eql_installed(pool, super::EQL_INSTALL_SQL).await?;
+    // CI shard it has no `eql_v3` surface, so apply the migrations (idempotent +
+    // process-safe via the migrator's advisory lock) before any cast/query.
+    ensure_eql_installed(pool, &super::migrator()).await?;
     ensure_fixture_loaded::<T>(pool, embedded_fixture_sql::<T>()).await?;
     let sql = format!(
         "SELECT plaintext, payload::text FROM {} ORDER BY id",

tests/sqlx/tests/encrypted_domain/property/mod.rs

@@ -4,18 +4,18 @@
 //! `scalars::<X>::` test-name prefix, so a `scalars::property::…` test would be
 //! mis-read as a scalar type and break the catalog cross-check.
 
-/// The EQL installer (`migrations/001_install_eql.sql`, the full release),
-/// embedded at compile time so the property suites can install the `eql_v3`
-/// surface into their base DB on demand (see `property::ensure_eql_installed`).
-/// Same embed-into-the-archive rationale as the per-type fixture corpus in
-/// `fixture_oracle.rs`: the file is produced by `test:sqlx:prep` before the
-/// nextest archive is built, and the CI shards run from that archive without a
-/// checkout of the gitignored generated SQL. The path resolves against the
-/// `eql_tests` crate root (`tests/sqlx`).
-pub(crate) const EQL_INSTALL_SQL: &str = include_str!(concat!(
-    env!("CARGO_MANIFEST_DIR"),
-    "/migrations/001_install_eql.sql"
-));
+/// The embedded SQLx migration set (`tests/sqlx/migrations`) — the SAME one
+/// `#[sqlx::test]` applies to its scratch DBs. The property suites connect to
+/// the base DB directly (their proptest case loop is sync and can't take
+/// `#[sqlx::test]`'s injected pool), so they apply this themselves to reach the
+/// migrated state the rest of the suite gets for free (see
+/// `property::ensure_eql_installed`). The macro embeds the files at compile
+/// time and resolves `./migrations` against the `eql_tests` crate root
+/// (`tests/sqlx`); kept in the test target, not the lib, so the lib never
+/// embeds the gitignored generated `001_install_eql.sql`.
+pub(crate) fn migrator() -> sqlx::migrate::Migrator {
+    sqlx::migrate!("./migrations")
+}
 
 // NULL / blocker / CHECK-constraint unit tests.
 mod edge_cases;