fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: give the DDL and global-count suites their own database

Browse Source 
Turning CI on surfaced `atomic_write` as intermittently failing under load —
self-inflicted. Its fault injection is `CREATE TRIGGER ... ON outbox`, which takes
an ACCESS EXCLUSIVE lock on a table every other suite is concurrently inserting
into. The header even claimed it "cannot affect any test running in parallel";
that was wrong — scoping the trigger by app_id bounds which ROWS it rejects, not
the table lock installing it takes. Shipping that alongside "run the DB tests in
CI" would have poisoned the signal.

The fix is the one already used for the e2e suites: a private database per test.
That logic (build a migrated template once, clone it per test via
`CREATE DATABASE ... TEMPLATE`) was duplicated in `picloud/tests/common`, and this
would have been a third copy — so it moves into a shared `picloud-test-support`
dev-dependency crate, and `picloud/tests/common` now re-exports it. The invariant
it encodes: manager-core's claim loops are global by design, so a test must not
share a database with anything that runs them OR with anything that does DDL.

Moved onto it:
- `atomic_write` — the DDL fault injection above; its per-test cleanup is now
  deleted (the database is thrown away).
- `outbox_reclaim` — asserts `reclaim_stale_claims` returns exactly 0, a count
  over the WHOLE outbox table. On the shared DB any stale row from any other suite
  (or a killed picloud that died holding a claim) would fail it permanently, until
  someone cleared the table by hand.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-15 07:27:35 +02:00
parent b5ce4cec01
commit 612074af04
9 changed files with 262 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
Cargo.lock generated
View File

@@ -1939,6 +1939,7 @@ dependencies = [
"picloud-manager-core", "picloud-manager-core",
"picloud-orchestrator-core", "picloud-orchestrator-core",
"picloud-shared", "picloud-shared",
"picloud-test-support",
"serde", "serde",
"serde_json", "serde_json",
"sha2 0.10.9", "sha2 0.10.9",
@@ -2043,6 +2044,7 @@ dependencies = [
"picloud-executor-core", "picloud-executor-core",
"picloud-orchestrator-core", "picloud-orchestrator-core",
"picloud-shared", "picloud-shared",
"picloud-test-support",
"rand 0.8.6", "rand 0.8.6",
"reqwest", "reqwest",
"serde", "serde",
@@ -2110,6 +2112,14 @@ dependencies = [
"uuid", "uuid",
] ]
[[package]]
name = "picloud-test-support"
version = "1.1.9"
dependencies = [
"sqlx",
"tokio",
]
[[package]] [[package]]
name = "pin-project-lite" name = "pin-project-lite"
version = "0.2.17" version = "0.2.17"

2
Cargo.toml
View File

@@ -10,6 +10,7 @@ members = [
"crates/picloud-orchestrator", "crates/picloud-orchestrator",
"crates/picloud-executor", "crates/picloud-executor",
"crates/picloud-cli", "crates/picloud-cli",
"crates/test-support",
] ]
[workspace.package] [workspace.package]
@@ -63,6 +64,7 @@ futures = "0.3"
rhai = { version = "=1.24", features = ["sync", "serde"] } rhai = { version = "=1.24", features = ["sync", "serde"] }
# Postgres (manager-core only — others stay DB-free) # Postgres (manager-core only — others stay DB-free)
picloud-test-support = { path = "crates/test-support" }
sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "json", "macros", "migrate"] } sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "json", "macros", "migrate"] }
# Config # Config

1
crates/manager-core/Cargo.toml
View File

@@ -41,4 +41,5 @@ data-encoding.workspace = true
lettre.workspace = true lettre.workspace = true
[dev-dependencies] [dev-dependencies]
picloud-test-support.workspace = true
tokio.workspace = true tokio.workspace = true

49
crates/manager-core/tests/atomic_write.rs
View File

@@ -9,9 +9,15 @@
//! an emit failure rolls the write back and surfaces as an error. //! an emit failure rolls the write back and surfaces as an error.
//! //!
//! Fault injection: a Postgres BEFORE-INSERT trigger on `outbox` that raises, //! Fault injection: a Postgres BEFORE-INSERT trigger on `outbox` that raises,
//! scoped to THIS test's freshly-minted `app_id`, so it cannot affect any test //! scoped to this test's freshly-minted `app_id`.
//! running in parallel against the same database. Skips when `DATABASE_URL` is //!
//! unset. //! That scoping bounds which ROWS the trigger rejects, but installing it is still
//! `CREATE TRIGGER ... ON outbox` — DDL, taking an ACCESS EXCLUSIVE lock on a
//! table every other suite is concurrently inserting into. On the shared dev
//! database that made this suite fail intermittently under load (and briefly
//! imposed our trigger on everyone else's inserts). So each test here now runs
//! against its OWN database, which is what makes the DDL safe; it also means no
//! test needs to clean up after itself. Skips when `DATABASE_URL` is unset.
use std::sync::Arc; use std::sync::Arc;
@@ -26,25 +32,13 @@ use picloud_shared::{
AppId, ExecutionId, FileUpdate, FilesError, GroupDocsError, GroupFilesError, GroupId, AppId, ExecutionId, FileUpdate, FilesError, GroupDocsError, GroupFilesError, GroupId,
GroupKvError, KvError, NewFile, RequestId, ScriptId, SdkCallCx, GroupKvError, KvError, NewFile, RequestId, ScriptId, SdkCallCx,
}; };
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool; use sqlx::PgPool;
use uuid::Uuid; use uuid::Uuid;
async fn pool_or_skip() -> Option<PgPool> { async fn pool_or_skip() -> Option<PgPool> {
let Ok(url) = std::env::var("DATABASE_URL") else { // 24 connections: the quota races below spawn enough concurrent writers to
eprintln!("atomic_write: DATABASE_URL unset — skipping"); // actually contend for the per-group advisory lock.
return None; picloud_test_support::test_pool_sized("atomic_write", 24).await
};
let pool = PgPoolOptions::new()
.max_connections(24)
.connect(&url)
.await
.expect("connect");
sqlx::migrate!("./migrations")
.run(&pool)
.await
.expect("migrate");
Some(pool)
} }
fn cx(app_id: Uuid) -> SdkCallCx { fn cx(app_id: Uuid) -> SdkCallCx {
@@ -175,19 +169,6 @@ async fn outbox_count(pool: &PgPool, app: Uuid) -> i64 {
} }
/// `scripts` is RESTRICT on app delete (code is not data), so drop it first. /// `scripts` is RESTRICT on app delete (code is not data), so drop it first.
async fn cleanup(pool: &PgPool, app: Uuid) {
sqlx::query("DELETE FROM scripts WHERE app_id = $1")
.bind(app)
.execute(pool)
.await
.expect("cleanup scripts");
sqlx::query("DELETE FROM apps WHERE id = $1")
.bind(app)
.execute(pool)
.await
.expect("cleanup app");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_failed_fan_out_rolls_the_kv_write_back() { async fn a_failed_fan_out_rolls_the_kv_write_back() {
let Some(pool) = pool_or_skip().await else { let Some(pool) = pool_or_skip().await else {
@@ -243,8 +224,6 @@ async fn a_failed_fan_out_rolls_the_kv_write_back() {
.expect("a write with no matching trigger needs no outbox row"); .expect("a write with no matching trigger needs no outbox row");
assert_eq!(kv_count(&pool, f.app).await, 2); assert_eq!(kv_count(&pool, f.app).await, 2);
assert_eq!(outbox_count(&pool, f.app).await, 1, "still no new fan-out"); assert_eq!(outbox_count(&pool, f.app).await, 1, "still no new fan-out");
cleanup(&pool, f.app).await;
} }
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -277,8 +256,6 @@ async fn a_failed_fan_out_rolls_a_delete_back() {
1, 1,
"the delete rolled back — the key is still there" "the delete rolled back — the key is still there"
); );
cleanup(&pool, f.app).await;
} }
// ---------------------------------------------------------------------------- // ----------------------------------------------------------------------------
@@ -657,7 +634,6 @@ async fn an_app_cannot_exceed_its_key_ceiling_but_updates_stay_free() {
.expect("room after a delete"); .expect("room after a delete");
assert_eq!(kv_count(&pool, f.app).await, 3); assert_eq!(kv_count(&pool, f.app).await, 3);
cleanup(&pool, f.app).await;
} }
#[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
@@ -759,6 +735,5 @@ async fn concurrent_uploads_cannot_push_an_app_past_its_disk_ceiling() {
"the original bytes must be intact — a refused update must not touch the blob" "the original bytes must be intact — a refused update must not touch the blob"
); );
cleanup(&pool, f.app).await;
let _ = std::fs::remove_dir_all(&root); let _ = std::fs::remove_dir_all(&root);
} }

27
crates/manager-core/tests/outbox_reclaim.rs
View File

@@ -17,25 +17,16 @@ use picloud_manager_core::outbox_repo::{
NewOutboxRow, OutboxRepo, OutboxSourceKind, PostgresOutboxRepo, NewOutboxRow, OutboxRepo, OutboxSourceKind, PostgresOutboxRepo,
}; };
use picloud_shared::AppId; use picloud_shared::AppId;
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool; use sqlx::PgPool;
use uuid::Uuid; use uuid::Uuid;
async fn pool_or_skip() -> Option<PgPool> { async fn pool_or_skip() -> Option<PgPool> {
let Ok(url) = std::env::var("DATABASE_URL") else { // A PRIVATE database. `reclaim_stale_claims` returns a count across the WHOLE
eprintln!("outbox_reclaim: DATABASE_URL unset — skipping"); // outbox, and the last assertion here demands that count be exactly 0 — on a
return None; // shared database any stale row left behind by any other suite (or by a
}; // killed picloud that died holding a claim) would fail it, permanently, until
let pool = PgPoolOptions::new() // someone cleared the table by hand.
.max_connections(3) picloud_test_support::test_pool("outbox_reclaim").await
.connect(&url)
.await
.expect("connect");
sqlx::migrate!("./migrations")
.run(&pool)
.await
.expect("migrate");
Some(pool)
} }
async fn mk_app(pool: &PgPool) -> Uuid { async fn mk_app(pool: &PgPool) -> Uuid {
@@ -144,10 +135,4 @@ async fn a_stranded_claim_is_reclaimed_without_burning_the_retry_budget() {
0, 0,
"a claim within the timeout belongs to a live dispatcher and must be left alone" "a claim within the timeout belongs to a live dispatcher and must be left alone"
); );
sqlx::query("DELETE FROM apps WHERE id = $1")
.bind(app)
.execute(&pool)
.await
.expect("cleanup");
} }

1
crates/picloud/Cargo.toml
View File

@@ -36,6 +36,7 @@ tracing-subscriber.workspace = true
figment.workspace = true figment.workspace = true
[dev-dependencies] [dev-dependencies]
picloud-test-support.workspace = true
axum-test = "17" axum-test = "17"
serde.workspace = true serde.workspace = true
serde_json.workspace = true serde_json.workspace = true

187
crates/picloud/tests/common/mod.rs
View File

@@ -1,184 +1,11 @@
//! A hermetic Postgres database per test. //! Per-test database for the e2e suites.
//! //!
//! ## Why this exists //! The implementation lives in `picloud-test-support` so `manager-core`'s
//! //! integration tests share one copy of this (subtle) template-clone logic rather
//! Every test in `dispatcher_e2e` / `queue_e2e` / `invoke_e2e` / `retry_e2e` / //! than a third hand-rolled duplicate. See that crate's docs for WHY each test
//! `email_inbound` calls `build_app`, which spawns a **real dispatcher**. When //! needs a private database — in short, the dispatcher's claim loops are global
//! those tests shared one database, they shared one `outbox` — and //! by design, so a test must not share a database with anything that runs them.
//! `OutboxRepo::claim_due` is deliberately NOT app-scoped (in production one
//! dispatcher serves the whole instance, so claiming any due row is correct).
//!
//! So test A's dispatcher would claim test B's outbox row, test A would finish,
//! its server would be dropped mid-dispatch, and the claim would be **stranded**.
//! Test B then polled for its handler's side effect until it timed out. The
//! result was a flake that moved around between runs — `dispatcher_e2e` failed
//! roughly one run in three, on a different test each time, and never when run
//! with `--test-threads=1`.
//!
//! Note what the harness was modelling: N independent *instances* sharing one
//! database. Production never does that — one instance owns one database. The
//! fix is therefore to make the tests match production rather than to weaken the
//! dispatcher (app-scoping `claim_due` would be wrong) or to shorten
//! `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` (600s is right for a data plane where a
//! script may legitimately run 300s; a 10s test window is not a reason to make
//! production abandon live claims).
//!
//! ## How
//!
//! Migrations are slow to replay 30+ times, so the first caller builds a
//! migrated **template** database once and every test then clones it with
//! `CREATE DATABASE ... TEMPLATE`, which is a file copy. Database names are
//! derived from (suite, test name), so they are stable: a test drops and
//! recreates its own database on entry. That means a crashed test leaves no
//! garbage to clean up — the next run reclaims the name — and there is no
//! cleanup step to forget.
//!
//! Skips cleanly (returns `None`) when `DATABASE_URL` is unset, matching the
//! existing `pool_or_skip` contract.
#![allow(dead_code)] #![allow(dead_code)]
use sqlx::postgres::PgPoolOptions; pub use picloud_test_support::test_pool;
use sqlx::{Connection, Executor, PgConnection, PgPool};
/// The migrated database every test database is cloned from.
const TEMPLATE_DB: &str = "picloud_test_template";
/// Serializes template creation and `CREATE DATABASE ... TEMPLATE` across every
/// concurrently-running test *process*. Advisory locks are cluster-wide, so this
/// works across test binaries, not just across threads. It is needed because
/// Postgres refuses to clone a template that has an open connection.
const TEMPLATE_LOCK: i64 = 0x0070_6963_6C6F_7564;
/// Split `postgres://…/picloud?sslmode=…` into (`everything before the db name`,
/// `db name`, `query suffix`) so we can address a sibling database on the same
/// server.
fn split_url(url: &str) -> Option<(String, String)> {
let (root, tail) = url.rsplit_once('/')?;
let (_db, query) = match tail.split_once('?') {
Some((db, q)) => (db, format!("?{q}")),
None => (tail, String::new()),
};
Some((root.to_string(), query))
}
/// Postgres identifiers cap at 63 bytes, so a long test name is truncated and
/// disambiguated by a hash of the full (suite, test) pair.
fn db_name(suite: &str, test: &str) -> String {
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
for b in suite
.bytes()
.chain(b"::".iter().copied())
.chain(test.bytes())
{
hash ^= u64::from(b);
hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
}
let safe: String = test
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() {
c.to_ascii_lowercase()
} else {
'_'
}
})
.take(38)
.collect();
format!("pt_{safe}_{:08x}", hash & 0xffff_ffff)
}
/// The running test's name. libtest names each test's thread after the test, and
/// `#[tokio::test]` drives the future on that same thread (`block_on` polls on the
/// caller), so this is stable for both the current-thread and multi-thread
/// flavors. Falls back to a constant if a future harness ever stops naming them —
/// tests would then share a database again, which is exactly today's behavior, so
/// the fallback cannot be worse than the status quo.
fn test_name() -> String {
std::thread::current()
.name()
.unwrap_or("unnamed")
.rsplit("::")
.next()
.unwrap_or("unnamed")
.to_string()
}
/// Connect to a database private to this test, or `None` when `DATABASE_URL` is
/// unset. `suite` disambiguates same-named tests in different binaries.
pub async fn test_pool(suite: &str) -> Option<PgPool> {
let Ok(url) = std::env::var("DATABASE_URL") else {
eprintln!("{suite}: DATABASE_URL unset — skipping");
return None;
};
let (root, query) = split_url(&url).expect("DATABASE_URL must name a database");
let name = db_name(suite, &test_name());
let mut admin = PgConnection::connect(&url)
.await
.expect("connect to DATABASE_URL");
// Hold the lock across BOTH the template build and the clone: Postgres will
// not clone a template that has an open connection, and the migrating
// connection below is exactly such a connection.
sqlx::query("SELECT pg_advisory_lock($1)")
.bind(TEMPLATE_LOCK)
.execute(&mut admin)
.await
.expect("take the template lock");
ensure_template(&mut admin, &root, &query).await;
// A stable name means a crashed test leaves nothing behind: the next run
// simply reclaims it. FORCE evicts any connection a previous run leaked.
admin
.execute(format!(r#"DROP DATABASE IF EXISTS "{name}" WITH (FORCE)"#).as_str())
.await
.expect("drop the previous test database");
admin
.execute(format!(r#"CREATE DATABASE "{name}" TEMPLATE "{TEMPLATE_DB}""#).as_str())
.await
.expect("clone the template");
sqlx::query("SELECT pg_advisory_unlock($1)")
.bind(TEMPLATE_LOCK)
.execute(&mut admin)
.await
.expect("release the template lock");
admin.close().await.ok();
let pool = PgPoolOptions::new()
.max_connections(5)
.connect(&format!("{root}/{name}{query}"))
.await
.expect("connect to the test database");
Some(pool)
}
/// Create the template if absent, then migrate it. Migrating on every call (not
/// just on creation) is what picks up a newly-added migration; it is a handful of
/// queries once the template is current. The caller must hold [`TEMPLATE_LOCK`],
/// and this must leave no connection open to the template.
async fn ensure_template(admin: &mut PgConnection, root: &str, query: &str) {
let exists: Option<(i32,)> = sqlx::query_as("SELECT 1 FROM pg_database WHERE datname = $1")
.bind(TEMPLATE_DB)
.fetch_optional(&mut *admin)
.await
.expect("look up the template");
if exists.is_none() {
admin
.execute(format!(r#"CREATE DATABASE "{TEMPLATE_DB}""#).as_str())
.await
.expect("create the template");
}
let mut tmpl = PgConnection::connect(&format!("{root}/{TEMPLATE_DB}{query}"))
.await
.expect("connect to the template");
sqlx::migrate!("../manager-core/migrations")
.run(&mut tmpl)
.await
.expect("migrate the template");
// Must be closed before the clone — see the lock comment above.
tmpl.close().await.expect("close the template connection");
}

13
crates/test-support/Cargo.toml Normal file
View File

@@ -0,0 +1,13 @@
[package]
name = "picloud-test-support"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
publish = false
# Test-only. A dev-dependency of the crates whose integration tests need a
# private database; never a runtime dependency of anything shipped.
[dependencies]
sqlx.workspace = true
tokio.workspace = true

210
crates/test-support/src/lib.rs Normal file
View File

@@ -0,0 +1,210 @@
//! A hermetic Postgres database per test.
//!
//! ## The invariant
//!
//! **`manager-core`'s claim loops are global by design, so a test must not share
//! a database with anything that runs them — or with anything that does DDL.**
//!
//! In production one instance owns one database, and `OutboxRepo::claim_due`,
//! `claim_ready_step` and the reclaimers deliberately take no `app_id`: one
//! dispatcher serves every app. The test suites were the only place that ever
//! violated the "one instance, one database" assumption, by pointing N
//! independent instances at one database. Two concrete bugs came from it:
//!
//! * `picloud/tests/*_e2e` — each test spawned a real dispatcher via `build_app`,
//! so test A's dispatcher claimed test B's outbox row, then A finished and its
//! server dropped mid-dispatch, stranding the claim. B polled until it timed
//! out. `dispatcher_e2e` failed ~1 run in 3, never under `--test-threads=1`.
//!
//! * `manager-core/tests/atomic_write` — its fault injection does
//! `CREATE TRIGGER ... ON outbox`, which takes an ACCESS EXCLUSIVE lock on a
//! table other suites are concurrently inserting into. Intermittent failures
//! under load, and the DDL was visible to every other suite while installed.
//!
//! The fix in both cases is the same, and it is the harness's job, not the
//! product's: do not weaken the dispatcher (app-scoping `claim_due` would be
//! wrong), do not shorten `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` to fit a 10s test
//! window (600s is right for a data plane where a script may run 300s). Give each
//! test its own database, so the harness matches production.
//!
//! ## How
//!
//! Replaying the migrations per test is far too slow, so the first caller builds a
//! migrated **template** database once and every test then clones it with
//! `CREATE DATABASE ... TEMPLATE`, which is a file copy. Names derive from
//! (suite, test), so they are stable: a test drops and recreates its own database
//! on entry. A crashed test therefore leaves nothing to clean up — the next run
//! reclaims the name — and there is no teardown to forget. Tests using this need
//! no `DELETE FROM` cleanup at all.
//!
//! Returns `None` when `DATABASE_URL` is unset, matching the existing
//! skip-cleanly-without-a-DB convention.
use sqlx::postgres::PgPoolOptions;
use sqlx::{Connection, Executor, PgConnection, PgPool};
/// The migrated database every test database is cloned from.
const TEMPLATE_DB: &str = "picloud_test_template";
/// Serializes template creation and `CREATE DATABASE ... TEMPLATE` across every
/// concurrently-running test *process*. Advisory locks are cluster-wide, so this
/// works across test binaries, not just across threads. It is needed because
/// Postgres refuses to clone a template that has an open connection.
const TEMPLATE_LOCK: i64 = 0x0070_6963_6C6F_7564;
/// Split a URL into (everything before the database name, query suffix) so we can
/// address a sibling database on the same server.
fn split_url(url: &str) -> Option<(String, String)> {
let (root, tail) = url.rsplit_once('/')?;
let query = match tail.split_once('?') {
Some((_, q)) => format!("?{q}"),
None => String::new(),
};
Some((root.to_string(), query))
}
/// Postgres identifiers cap at 63 bytes, so a long test name is truncated and
/// disambiguated by a hash of the full (suite, test) pair.
fn db_name(suite: &str, test: &str) -> String {
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
for b in suite
.bytes()
.chain(b"::".iter().copied())
.chain(test.bytes())
{
hash ^= u64::from(b);
hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
}
let safe: String = test
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() {
c.to_ascii_lowercase()
} else {
'_'
}
})
.take(38)
.collect();
format!("pt_{safe}_{:08x}", hash & 0xffff_ffff)
}
/// The running test's name. libtest names each test's thread after the test, and
/// `#[tokio::test]` drives the future on that same thread (`block_on` polls on the
/// caller), so this is stable for both the current-thread and multi-thread
/// flavors. Falls back to a constant if a future harness stops naming them —
/// tests would then share a database again, which is merely the old behavior, so
/// the fallback cannot be worse than the status quo.
fn test_name() -> String {
std::thread::current()
.name()
.unwrap_or("unnamed")
.rsplit("::")
.next()
.unwrap_or("unnamed")
.to_string()
}
/// A database private to this test. `suite` disambiguates same-named tests in
/// different binaries. `None` when `DATABASE_URL` is unset.
///
/// # Panics
/// If `DATABASE_URL` is set but unusable (cannot connect, cannot create the
/// template, cannot clone it) — a misconfigured integration environment should
/// fail loudly, not silently skip.
pub async fn test_db_url(suite: &str) -> Option<String> {
let Ok(url) = std::env::var("DATABASE_URL") else {
eprintln!("{suite}: DATABASE_URL unset — skipping");
return None;
};
let (root, query) = split_url(&url).expect("DATABASE_URL must name a database");
let name = db_name(suite, &test_name());
let mut admin = PgConnection::connect(&url)
.await
.expect("connect to DATABASE_URL");
// Held across BOTH the template build and the clone: Postgres will not clone a
// template that has an open connection, and the migrating connection below is
// exactly such a connection.
sqlx::query("SELECT pg_advisory_lock($1)")
.bind(TEMPLATE_LOCK)
.execute(&mut admin)
.await
.expect("take the template lock");
ensure_template(&mut admin, &root, &query).await;
// A stable name means a crashed test leaves nothing behind: the next run
// reclaims it. FORCE evicts any connection a previous run leaked.
admin
.execute(format!(r#"DROP DATABASE IF EXISTS "{name}" WITH (FORCE)"#).as_str())
.await
.expect("drop the previous test database");
admin
.execute(format!(r#"CREATE DATABASE "{name}" TEMPLATE "{TEMPLATE_DB}""#).as_str())
.await
.expect("clone the template");
sqlx::query("SELECT pg_advisory_unlock($1)")
.bind(TEMPLATE_LOCK)
.execute(&mut admin)
.await
.expect("release the template lock");
admin.close().await.ok();
Some(format!("{root}/{name}{query}"))
}
/// A pool onto a database private to this test. See [`test_db_url`].
///
/// # Panics
/// If the private database cannot be created or connected to.
pub async fn test_pool(suite: &str) -> Option<PgPool> {
test_pool_sized(suite, 5).await
}
/// [`test_pool`] with an explicit connection cap — for suites that drive many
/// concurrent writers at once (e.g. the quota races in `atomic_write`, which need
/// enough connections to actually contend for the advisory lock).
///
/// # Panics
/// If the private database cannot be created or connected to.
pub async fn test_pool_sized(suite: &str, max_connections: u32) -> Option<PgPool> {
let url = test_db_url(suite).await?;
Some(
PgPoolOptions::new()
.max_connections(max_connections)
.connect(&url)
.await
.expect("connect to the test database"),
)
}
/// Create the template if absent, then migrate it. Migrating on every call (not
/// only on creation) is what picks up a newly-added migration; it is a handful of
/// queries once the template is current. The caller must hold [`TEMPLATE_LOCK`],
/// and this must leave no connection open to the template.
async fn ensure_template(admin: &mut PgConnection, root: &str, query: &str) {
let exists: Option<(i32,)> = sqlx::query_as("SELECT 1 FROM pg_database WHERE datname = $1")
.bind(TEMPLATE_DB)
.fetch_optional(&mut *admin)
.await
.expect("look up the template");
if exists.is_none() {
admin
.execute(format!(r#"CREATE DATABASE "{TEMPLATE_DB}""#).as_str())
.await
.expect("create the template");
}
let mut tmpl = PgConnection::connect(&format!("{root}/{TEMPLATE_DB}{query}"))
.await
.expect("connect to the template");
sqlx::migrate!("../manager-core/migrations")
.run(&mut tmpl)
.await
.expect("migrate the template");
// Must be closed before the clone — see the lock comment above.
tmpl.close().await.expect("close the template connection");
}