diff --git a/Cargo.lock b/Cargo.lock index ce51ffb..57512bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1939,6 +1939,7 @@ dependencies = [ "picloud-manager-core", "picloud-orchestrator-core", "picloud-shared", + "picloud-test-support", "serde", "serde_json", "sha2 0.10.9", @@ -2043,6 +2044,7 @@ dependencies = [ "picloud-executor-core", "picloud-orchestrator-core", "picloud-shared", + "picloud-test-support", "rand 0.8.6", "reqwest", "serde", @@ -2110,6 +2112,14 @@ dependencies = [ "uuid", ] +[[package]] +name = "picloud-test-support" +version = "1.1.9" +dependencies = [ + "sqlx", + "tokio", +] + [[package]] name = "pin-project-lite" version = "0.2.17" diff --git a/Cargo.toml b/Cargo.toml index 7153515..6134560 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ members = [ "crates/picloud-orchestrator", "crates/picloud-executor", "crates/picloud-cli", + "crates/test-support", ] [workspace.package] @@ -63,6 +64,7 @@ futures = "0.3" rhai = { version = "=1.24", features = ["sync", "serde"] } # 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"] } # Config diff --git a/crates/manager-core/Cargo.toml b/crates/manager-core/Cargo.toml index eabe31f..d482497 100644 --- a/crates/manager-core/Cargo.toml +++ b/crates/manager-core/Cargo.toml @@ -41,4 +41,5 @@ data-encoding.workspace = true lettre.workspace = true [dev-dependencies] +picloud-test-support.workspace = true tokio.workspace = true diff --git a/crates/manager-core/tests/atomic_write.rs b/crates/manager-core/tests/atomic_write.rs index c98701b..f51301c 100644 --- a/crates/manager-core/tests/atomic_write.rs +++ b/crates/manager-core/tests/atomic_write.rs @@ -9,9 +9,15 @@ //! an emit failure rolls the write back and surfaces as an error. //! //! 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 -//! running in parallel against the same database. Skips when `DATABASE_URL` is -//! unset. +//! scoped to this test's freshly-minted `app_id`. +//! +//! 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; @@ -26,25 +32,13 @@ use picloud_shared::{ AppId, ExecutionId, FileUpdate, FilesError, GroupDocsError, GroupFilesError, GroupId, GroupKvError, KvError, NewFile, RequestId, ScriptId, SdkCallCx, }; -use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; use uuid::Uuid; async fn pool_or_skip() -> Option { - let Ok(url) = std::env::var("DATABASE_URL") else { - eprintln!("atomic_write: DATABASE_URL unset — skipping"); - return None; - }; - let pool = PgPoolOptions::new() - .max_connections(24) - .connect(&url) - .await - .expect("connect"); - sqlx::migrate!("./migrations") - .run(&pool) - .await - .expect("migrate"); - Some(pool) + // 24 connections: the quota races below spawn enough concurrent writers to + // actually contend for the per-group advisory lock. + picloud_test_support::test_pool_sized("atomic_write", 24).await } 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. -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)] async fn a_failed_fan_out_rolls_the_kv_write_back() { 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"); assert_eq!(kv_count(&pool, f.app).await, 2); 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)] @@ -277,8 +256,6 @@ async fn a_failed_fan_out_rolls_a_delete_back() { 1, "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"); assert_eq!(kv_count(&pool, f.app).await, 3); - cleanup(&pool, f.app).await; } #[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" ); - cleanup(&pool, f.app).await; let _ = std::fs::remove_dir_all(&root); } diff --git a/crates/manager-core/tests/outbox_reclaim.rs b/crates/manager-core/tests/outbox_reclaim.rs index 5f1c9a2..544bec1 100644 --- a/crates/manager-core/tests/outbox_reclaim.rs +++ b/crates/manager-core/tests/outbox_reclaim.rs @@ -17,25 +17,16 @@ use picloud_manager_core::outbox_repo::{ NewOutboxRow, OutboxRepo, OutboxSourceKind, PostgresOutboxRepo, }; use picloud_shared::AppId; -use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; use uuid::Uuid; async fn pool_or_skip() -> Option { - let Ok(url) = std::env::var("DATABASE_URL") else { - eprintln!("outbox_reclaim: DATABASE_URL unset — skipping"); - return None; - }; - let pool = PgPoolOptions::new() - .max_connections(3) - .connect(&url) - .await - .expect("connect"); - sqlx::migrate!("./migrations") - .run(&pool) - .await - .expect("migrate"); - Some(pool) + // A PRIVATE database. `reclaim_stale_claims` returns a count across the WHOLE + // outbox, and the last assertion here demands that count be exactly 0 — on a + // 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 + // someone cleared the table by hand. + picloud_test_support::test_pool("outbox_reclaim").await } async fn mk_app(pool: &PgPool) -> Uuid { @@ -144,10 +135,4 @@ async fn a_stranded_claim_is_reclaimed_without_burning_the_retry_budget() { 0, "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"); } diff --git a/crates/picloud/Cargo.toml b/crates/picloud/Cargo.toml index a9297ca..9c441c2 100644 --- a/crates/picloud/Cargo.toml +++ b/crates/picloud/Cargo.toml @@ -36,6 +36,7 @@ tracing-subscriber.workspace = true figment.workspace = true [dev-dependencies] +picloud-test-support.workspace = true axum-test = "17" serde.workspace = true serde_json.workspace = true diff --git a/crates/picloud/tests/common/mod.rs b/crates/picloud/tests/common/mod.rs index 1b86970..66051fd 100644 --- a/crates/picloud/tests/common/mod.rs +++ b/crates/picloud/tests/common/mod.rs @@ -1,184 +1,11 @@ -//! A hermetic Postgres database per test. +//! Per-test database for the e2e suites. //! -//! ## Why this exists -//! -//! Every test in `dispatcher_e2e` / `queue_e2e` / `invoke_e2e` / `retry_e2e` / -//! `email_inbound` calls `build_app`, which spawns a **real dispatcher**. When -//! those tests shared one database, they shared one `outbox` — and -//! `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. +//! The implementation lives in `picloud-test-support` so `manager-core`'s +//! integration tests share one copy of this (subtle) template-clone logic rather +//! than a third hand-rolled duplicate. See that crate's docs for WHY each test +//! needs a private database — in short, the dispatcher's claim loops are global +//! by design, so a test must not share a database with anything that runs them. #![allow(dead_code)] -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 `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 { - 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"); -} +pub use picloud_test_support::test_pool; diff --git a/crates/test-support/Cargo.toml b/crates/test-support/Cargo.toml new file mode 100644 index 0000000..4e990b1 --- /dev/null +++ b/crates/test-support/Cargo.toml @@ -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 diff --git a/crates/test-support/src/lib.rs b/crates/test-support/src/lib.rs new file mode 100644 index 0000000..09e5f87 --- /dev/null +++ b/crates/test-support/src/lib.rs @@ -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 { + 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 { + 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 { + 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"); +}