jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(buzz-db): drop partition scratch schemas on panic (#8293)

Browse Source 
## Summary

The partition Postgres tests dropped their scratch schema with a
`drop_schema` call at the end of each test body. When a test panicked,
that call never ran, so the schema leaked into the shared database,
along with any `{schema}_*` sibling schemas or reader role the test
created. A reviewer on #7993 hit this locally: the leaked schema plus a
database-wide catalog query made later runs fail. CI hides it because
the test wrapper gives each process its own database.

`scratch_pool` / `scratch_pool_with_max_connections` now return `(pool,
ScratchSchema)`. The guard owns the admin pool and schema name and
cleans up on `Drop`:

- **Fresh thread and connection.** `Drop` runs on the test's runtime
thread, which can't drive the test's pools while it blocks, so cleanup
runs on its own thread with its own runtime and connection.
- **Terminates the test's backends first.** Both pools record each
connection's backend pid. A test that panics mid-transaction keeps its
locks, because the rollback would only run on the blocked runtime.
Without this step `DROP SCHEMA` waits on those locks.
- **Drops `{name}` plus every schema and role named `{name}_*`.** This
replaces the per-test sibling-schema and `DROP ROLE` cleanup. The reader
role is now named `{schema}_reader`.
- **Bounded.** `lock_timeout = '10s'`. Failures print a warning and
never panic inside `Drop`.

The 52 call sites are migrated mechanically. Test-only change.

### Related issue

Follow-up from review on #7993. No existing issue found.

### Testing

- New `scratch_schema_is_dropped_when_a_test_panics_holding_a_lock`: the
test body creates a sibling schema, takes `ACCESS EXCLUSIVE` on a
scratch table in an open transaction, and panics. The test then asserts
no `{name}` / `{name}_*` schema remains.
- Proved it can fail: with the backend-termination step disabled,
cleanup hits the lock timeout and the test fails, reporting both leaked
schemas.
- Partition Postgres lane: `scripts/postgres-test-run.sh -p buzz-db
--lib -E 'test(partition)'`, 53/53 passed.
- The same filter run directly against the shared `buzz` database: 53/53
passed, then 0 `partition_audit_test_%` schemas and 0
`partition_audit_%` roles remained.
- `cargo clippy -p buzz-db --tests` clean. Pre-push hooks passed.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Ian Oberst <ioberst@block.xyz>
Co-authored-by: Ian Oberst <ioberst@block.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ian Oberst
2026-10-14 11:36:26 -07:00
committed by GitHub
parent 99c5fc8bba
commit 2270b66e14
2 changed files with 341 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

418
crates/buzz-db/src/store/partition.rs
View File

@@ -2201,49 +2201,207 @@ mod tests {
}
pub(super) mod postgres_tests {
use std::sync::{Arc, Mutex};
use sqlx::postgres::PgPoolOptions;
use sqlx::Connection;
use uuid::Uuid;
use super::*;
pub(in crate::store::partition) async fn scratch_pool() -> (PgPool, PgPool, String) {
/// A uniquely named scratch schema that is dropped when the guard goes
/// out of scope, including when the test panics.
///
/// Cleanup also drops every schema and role named `{name}_*`, so a test
/// that needs extra schemas or roles derives their names from `name`.
///
/// Bind the guard to a named variable such as `_scratch` for the whole
/// test: `let (pool, _) = scratch_pool().await` drops it, and the
/// schema with it, immediately.
pub(in crate::store::partition) struct ScratchSchema {
pub(in crate::store::partition) admin: PgPool,
pub(in crate::store::partition) name: String,
url: String,
backends: Arc<Mutex<Vec<i32>>>,
}
impl Drop for ScratchSchema {
fn drop(&mut self) {
// Drop runs on the test's runtime thread, which cannot drive the
// test's pools while it blocks here, so clean up over a fresh
// connection on a thread with its own runtime.
let url = std::mem::take(&mut self.url);
let name = self.name.clone();
let backends = std::mem::take(
&mut *self
.backends
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
);
let cleanup = std::thread::spawn(move || {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| error.to_string())?
.block_on(drop_scratch_objects(&url, &name, &backends))
})
.join();
let error = match cleanup {
Ok(Ok(())) => return,
Ok(Err(error)) => error,
Err(_) => "cleanup thread panicked".to_owned(),
};
eprintln!(
"warning: failed to drop scratch schema {}: {error}",
self.name
);
}
}
async fn drop_scratch_objects(
url: &str,
name: &str,
backends: &[i32],
) -> std::result::Result<(), String> {
let mut connection = PgConnection::connect(url)
.await
.map_err(|error| error.to_string())?;
// Every step is idempotent, so keep going after a failure and
// report all of them rather than skipping the remaining objects.
let mut errors = Vec::new();
let mut record = |step: &str, result: sqlx::Result<()>| {
if let Err(error) = result {
errors.push(format!("{step}: {error}"));
}
};
let result = sqlx::query("SET lock_timeout = '10s'")
.execute(&mut connection)
.await
.map(drop);
record("set lock_timeout", result);
// A panicking test can leave a transaction holding locks in the
// schema; its rollback would only run on the blocked test runtime.
let result = sqlx::query(
"SELECT pg_terminate_backend(pid, 5000) FROM pg_stat_activity \
WHERE pid = ANY($1) AND datname = current_database()",
)
.bind(backends)
.execute(&mut connection)
.await
.map(drop);
record("terminate test backends", result);
let schemas: Vec<String> = sqlx::query_scalar(
"SELECT nspname::text FROM pg_namespace \
WHERE nspname = $1 OR starts_with(nspname, $1 || '_')",
)
.bind(name)
.fetch_all(&mut connection)
.await
.unwrap_or_else(|error| {
record("list schemas", Err(error));
Vec::new()
});
for schema in schemas {
let result = sqlx::query(sqlx::AssertSqlSafe(format!(
"DROP SCHEMA IF EXISTS {} CASCADE",
quote_identifier(&schema)
)))
.execute(&mut connection)
.await
.map(drop);
record(&format!("drop schema {schema}"), result);
}
let roles: Vec<String> = sqlx::query_scalar(
"SELECT rolname::text FROM pg_roles WHERE starts_with(rolname, $1 || '_')",
)
.bind(name)
.fetch_all(&mut connection)
.await
.unwrap_or_else(|error| {
record("list roles", Err(error));
Vec::new()
});
for role in roles {
let result = sqlx::query(sqlx::AssertSqlSafe(format!(
"DROP ROLE IF EXISTS {}",
quote_identifier(&role)
)))
.execute(&mut connection)
.await
.map(drop);
record(&format!("drop role {role}"), result);
}
record("close connection", connection.close().await);
if errors.is_empty() {
Ok(())
} else {
Err(errors.join("; "))
}
}
/// Pool options that record each connection's backend pid in `backends`
/// and point its `search_path` at `search_path`, when given.
fn tracked_pool_options(
search_path: Option<String>,
backends: Arc<Mutex<Vec<i32>>>,
) -> PgPoolOptions {
PgPoolOptions::new().after_connect(move |connection, _| {
let search_path = search_path.clone();
let backends = Arc::clone(&backends);
Box::pin(async move {
if let Some(schema) = search_path {
sqlx::query(sqlx::AssertSqlSafe(format!("SET search_path TO {schema}")))
.execute(&mut *connection)
.await?;
}
let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
.fetch_one(&mut *connection)
.await?;
backends
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(pid);
Ok(())
})
})
}
/// A pool whose `search_path` is a fresh scratch schema, plus the guard
/// that drops it. Bind the guard to a named variable, never `_`.
pub(in crate::store::partition) async fn scratch_pool() -> (PgPool, ScratchSchema) {
scratch_pool_with_max_connections(4).await
}
pub(in crate::store::partition) async fn scratch_pool_with_max_connections(
max_connections: u32,
) -> (PgPool, PgPool, String) {
) -> (PgPool, ScratchSchema) {
let url = crate::test_support::database_url();
let schema = format!("partition_audit_test_{}", Uuid::new_v4().simple());
let admin = PgPool::connect(&url).await.expect("connect admin pool");
sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema}")))
.execute(&admin)
.await
.expect("create scratch schema");
let search_path_schema = schema.clone();
let pool = PgPoolOptions::new()
.max_connections(max_connections)
.after_connect(move |connection, _| {
let schema = search_path_schema.clone();
Box::pin(async move {
sqlx::query(sqlx::AssertSqlSafe(format!("SET search_path TO {schema}")))
.execute(connection)
.await?;
Ok(())
})
})
let name = format!("partition_audit_test_{}", Uuid::new_v4().simple());
let backends = Arc::new(Mutex::new(Vec::new()));
let admin = tracked_pool_options(None, Arc::clone(&backends))
.connect(&url)
.await
.expect("connect scratch pool");
(pool, admin, schema)
}
pub(in crate::store::partition) async fn drop_schema(admin: &PgPool, schema: &str) {
let _ = sqlx::query(sqlx::AssertSqlSafe(format!(
"DROP SCHEMA IF EXISTS {schema} CASCADE"
.expect("connect admin pool");
let scratch = ScratchSchema {
admin,
name,
url,
backends,
};
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE SCHEMA {}",
scratch.name
)))
.execute(admin)
.await;
.execute(&scratch.admin)
.await
.expect("create scratch schema");
let pool =
tracked_pool_options(Some(scratch.name.clone()), Arc::clone(&scratch.backends))
.max_connections(max_connections)
.connect(&scratch.url)
.await
.expect("connect scratch pool");
(pool, scratch)
}
pub(in crate::store::partition) async fn seed_parents(pool: &PgPool) {
@@ -2412,10 +2570,80 @@ mod tests {
}
}
#[test]
#[ignore = "requires Postgres"]
fn scratch_schema_is_dropped_when_a_test_panics_holding_a_lock() {
let runtime = || {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("build test runtime")
};
let (name_sender, name_receiver) = std::sync::mpsc::channel();
let outcome = std::thread::spawn(move || {
runtime().block_on(async move {
let (pool, scratch) = scratch_pool().await;
seed_parents(&pool).await;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE SCHEMA {}_children",
scratch.name
)))
.execute(&scratch.admin)
.await
.expect("create sibling schema");
for statement in [
format!("CREATE ROLE {}_reader NOLOGIN", scratch.name),
format!("GRANT USAGE ON SCHEMA {0} TO {0}_reader", scratch.name),
] {
sqlx::query(sqlx::AssertSqlSafe(statement))
.execute(&scratch.admin)
.await
.expect("create scratch role");
}
let mut holder = pool.begin().await.expect("begin lock holder");
sqlx::query("LOCK TABLE events IN ACCESS EXCLUSIVE MODE")
.execute(&mut *holder)
.await
.expect("lock scratch table");
name_sender
.send(scratch.name.clone())
.expect("send scratch schema name");
panic!("deliberate panic while holding a scratch schema lock");
})
})
.join();
assert!(outcome.is_err(), "the scratch test body must panic");
let name = name_receiver.recv().expect("scratch schema name");
let (schemas, roles): (Vec<String>, Vec<String>) = runtime().block_on(async {
let mut connection = PgConnection::connect(&crate::test_support::database_url())
.await
.expect("connect leftover check");
let schemas = sqlx::query_scalar(
"SELECT nspname::text FROM pg_namespace \
WHERE nspname = $1 OR starts_with(nspname, $1 || '_')",
)
.bind(&name)
.fetch_all(&mut connection)
.await
.expect("list leftover schemas");
let roles = sqlx::query_scalar(
"SELECT rolname::text FROM pg_roles WHERE starts_with(rolname, $1 || '_')",
)
.bind(&name)
.fetch_all(&mut connection)
.await
.expect("list leftover roles");
(schemas, roles)
});
assert!(schemas.is_empty(), "leaked scratch schemas: {schemas:?}");
assert!(roles.is_empty(), "leaked scratch roles: {roles:?}");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn fresh_layout_is_serving_safe_and_audit_is_read_only() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_fresh_layout(&pool).await;
let before = catalog_snapshot(&pool).await;
let audit = audit_partition_catalog_at(&pool, 3, fixed_now())
@@ -2431,13 +2659,12 @@ mod tests {
.all(|month| month.kind == MonthCoverageKind::CoveredByCatchAll)
}));
assert_eq!(catalog_snapshot(&pool).await, before);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn wrong_root_partition_key_never_proves_coverage() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents_with_keys(&pool, "alternate_at", "delivered_at").await;
for table in PARTITIONED_TABLES {
create_child(
@@ -2473,13 +2700,13 @@ mod tests {
.expect("delivery_log audit");
assert!(delivery_log.partition_key_valid);
assert!(delivery_log.serving_safe);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn occupied_catch_all_is_reported_as_degraded() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
let schema = scratch.name.as_str();
seed_fresh_layout(&pool).await;
sqlx::query(
"INSERT INTO events (created_at, alternate_at) \
@@ -2519,17 +2746,17 @@ mod tests {
assert!(delivery_log.children.iter().any(|child| {
child.name == "delivery_log_p_future" && child.catch_all_nonempty == Some(false)
}));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn occupied_cross_schema_catch_all_is_reported_as_degraded() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
let schema = scratch.name.as_str();
seed_fresh_layout(&pool).await;
let child_schema = format!("{schema}_children");
sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {child_schema}")))
.execute(&admin)
.execute(&scratch.admin)
.await
.expect("create child schema");
sqlx::query(sqlx::AssertSqlSafe(format!(
@@ -2567,20 +2794,19 @@ mod tests {
&& leaf.name == "events_p_future"
&& leaf.root_child_schema == child_schema
}));
drop_schema(&admin, &schema).await;
drop_schema(&admin, &child_schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn same_named_cross_schema_children_keep_distinct_trigger_metadata() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
let schema = scratch.name.as_str();
seed_parents(&pool).await;
let first_schema = format!("{schema}_first");
let second_schema = format!("{schema}_second");
for child_schema in [&first_schema, &second_schema] {
sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {child_schema}")))
.execute(&admin)
.execute(&scratch.admin)
.await
.expect("create child schema");
}
@@ -2676,19 +2902,17 @@ mod tests {
.expect("serialized coverage leaves")
.iter()
.any(|leaf| leaf["schema"] == second_schema));
drop_schema(&admin, &schema).await;
drop_schema(&admin, &first_schema).await;
drop_schema(&admin, &second_schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn report_retains_successful_table_when_another_table_errors() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
let schema = scratch.name.as_str();
seed_fresh_layout(&pool).await;
let role = format!("partition_audit_reader_{}", Uuid::new_v4().simple());
let role = format!("{schema}_reader");
sqlx::query(sqlx::AssertSqlSafe(format!("CREATE ROLE {role} NOLOGIN")))
.execute(&admin)
.execute(&scratch.admin)
.await
.expect("create restricted audit role");
for statement in [
@@ -2696,12 +2920,12 @@ mod tests {
format!("GRANT SELECT ON {schema}.events_p_future TO {role}"),
] {
sqlx::query(sqlx::AssertSqlSafe(statement))
.execute(&admin)
.execute(&scratch.admin)
.await
.expect("grant restricted audit access");
}
let url = crate::test_support::database_url();
let role_schema = schema.clone();
let role_schema = schema.to_owned();
let role_name = role.clone();
let restricted = PgPoolOptions::new()
.max_connections(1)
@@ -2731,17 +2955,12 @@ mod tests {
assert!(!report.serving_safe());
restricted.close().await;
drop_schema(&admin, &schema).await;
sqlx::query(sqlx::AssertSqlSafe(format!("DROP ROLE {role}")))
.execute(&admin)
.await
.expect("drop restricted audit role");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn pending_detach_never_proves_coverage() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_fresh_layout(&pool).await;
let mut blocker = pool.acquire().await.expect("acquire blocker");
@@ -2824,13 +3043,12 @@ mod tests {
&& child.pending_detach
&& child.kind == PartitionChildKind::Anomalous
}));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn repaired_layout_recognizes_bounds_not_catch_all_name() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
create_child(
@@ -2907,13 +3125,12 @@ mod tests {
assert!(august.tables.iter().all(PartitionTableAudit::degraded));
assert_eq!(august.outcome(), PartitionAuditOutcome::Degraded);
assert_eq!(catalog_snapshot(&pool).await, before);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn uncovered_months_are_created_with_trigger_parity() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
create_child(
@@ -2937,13 +3154,12 @@ mod tests {
.iter()
.all(|month| month.kind == MonthCoverageKind::CoveredByMonthly)
&& table.missing_trigger_count() == 0));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn default_partitions_report_occupancy_without_mutation() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_uncovered_layout(&pool).await;
for table in PARTITIONED_TABLES {
sqlx::query(sqlx::AssertSqlSafe(format!(
@@ -3037,13 +3253,12 @@ mod tests {
.await
.expect("check canonical relation"));
}
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn parent_default_does_not_mask_gap_below_explicit_sibling() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] {
create_child(
@@ -3116,13 +3331,13 @@ mod tests {
.execute(&pool)
.await
.is_err());
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn total_audit_timeout_bounds_blocked_occupancy_probes() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
let schema = scratch.name.as_str();
seed_uncovered_layout(&pool).await;
for table in PARTITIONED_TABLES {
sqlx::query(sqlx::AssertSqlSafe(format!(
@@ -3132,7 +3347,11 @@ mod tests {
.await
.expect("create default partition");
}
let mut blocker = admin.begin().await.expect("begin blocking transaction");
let mut blocker = scratch
.admin
.begin()
.await
.expect("begin blocking transaction");
sqlx::query(sqlx::AssertSqlSafe(format!(
"LOCK TABLE {schema}.events_default IN ACCESS EXCLUSIVE MODE"
)))
@@ -3157,13 +3376,12 @@ mod tests {
.all(|error| error.error.contains("exceeded total timeout of 100ms")));
blocker.rollback().await.expect("release blocking lock");
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn create_pins_non_iso_date_style() {
let (pool, admin, schema) = scratch_pool_with_max_connections(1).await;
let (pool, _scratch) = scratch_pool_with_max_connections(1).await;
seed_uncovered_layout(&pool).await;
sqlx::query("SET DateStyle TO 'SQL, DMY'")
.execute(&pool)
@@ -3186,13 +3404,12 @@ mod tests {
.await
.expect("show DateStyle");
assert_eq!(date_style, "SQL, DMY");
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn create_pins_non_utc_time_zone() {
let (pool, admin, schema) = scratch_pool_with_max_connections(1).await;
let (pool, _scratch) = scratch_pool_with_max_connections(1).await;
seed_uncovered_layout(&pool).await;
sqlx::query("SET TimeZone TO 'America/Los_Angeles'")
.execute(&pool)
@@ -3215,13 +3432,12 @@ mod tests {
.await
.expect("show TimeZone");
assert_eq!(time_zone, "America/Los_Angeles");
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn disabled_child_trigger_degrades_parity() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_fresh_layout(&pool).await;
sqlx::query("ALTER TABLE events_p_future DISABLE TRIGGER partition_probe")
.execute(&pool)
@@ -3240,13 +3456,12 @@ mod tests {
assert!(events.children.iter().any(|child| {
child.name == "events_p_future" && child.missing_triggers == ["partition_probe"]
}));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn always_enabled_parent_trigger_preserves_parity() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_fresh_layout(&pool).await;
sqlx::query("ALTER TABLE events ENABLE ALWAYS TRIGGER partition_probe")
.execute(&pool)
@@ -3266,13 +3481,13 @@ mod tests {
.children
.iter()
.all(|child| child.missing_triggers.is_empty()));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn disabled_nested_leaf_trigger_degrades_parity() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
let schema = scratch.name.as_str();
seed_parents(&pool).await;
for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] {
let nested = format!("{table}_nested");
@@ -3339,18 +3554,18 @@ mod tests {
assert_eq!(events.missing_trigger_count(), 1);
let expected = format!(
"{}:partition_probe",
qualified_relation_name(&schema, "events_nested_first")
qualified_relation_name(schema, "events_nested_first")
);
assert!(events.children.iter().any(|child| {
child.name == "events_nested" && child.missing_triggers == [expected.as_str()]
}));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn child_only_nested_leaf_trigger_degrades_parity() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
let schema = scratch.name.as_str();
seed_parents(&pool).await;
create_nested_child(
&pool,
@@ -3396,31 +3611,29 @@ mod tests {
assert_eq!(events.extra_trigger_count(), 1);
let expected = format!(
"{}:child_only_probe",
qualified_relation_name(&schema, "events_nested_first")
qualified_relation_name(schema, "events_nested_first")
);
assert!(events.children.iter().any(|child| {
child.name == "events_nested" && child.extra_triggers == [expected.as_str()]
}));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn catch_all_skips_create_without_catalog_mutation() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_fresh_layout(&pool).await;
let before = catalog_snapshot(&pool).await;
ensure_future_partitions_at(&pool, 3, true, fixed_now())
.await
.expect("covered no-op");
assert_eq!(catalog_snapshot(&pool).await, before);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn kill_switch_audits_but_does_not_create() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
create_child(
@@ -3445,13 +3658,12 @@ mod tests {
.months
.iter()
.all(|month| month.kind == MonthCoverageKind::Uncovered)));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn nested_children_without_leaves_do_not_prove_coverage() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] {
create_nested_child(
@@ -3477,13 +3689,12 @@ mod tests {
&& table.months[0].kind == MonthCoverageKind::Uncovered
}));
assert_eq!(catalog_snapshot(&pool).await, before);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn nested_descendant_leaves_prove_coverage_but_remain_degraded() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] {
let nested = format!("{table}_nested");
@@ -3519,13 +3730,12 @@ mod tests {
&& table.months[0].kind == MonthCoverageKind::CoveredByMonthly
}));
assert_eq!(catalog_snapshot(&pool).await, before);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn occupied_nested_catch_all_is_reported_on_leaf_and_root_child() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for (table, column) in [("events", "created_at"), ("delivery_log", "delivered_at")] {
let nested = format!("{table}_nested");
@@ -3584,13 +3794,12 @@ mod tests {
assert!(delivery_log.coverage_leaves.iter().any(|leaf| {
leaf.name == "delivery_log_nested_future" && leaf.catch_all_nonempty == Some(false)
}));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn nested_different_partition_key_does_not_prove_coverage() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
let nested = format!("{table}_nested");
@@ -3623,13 +3832,12 @@ mod tests {
&& table.coverage_leaves.is_empty()
&& table.months[0].kind == MonthCoverageKind::Uncovered
}));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn audit_pins_non_iso_session_rendering() {
let (pool, admin, schema) = scratch_pool_with_max_connections(1).await;
let (pool, _scratch) = scratch_pool_with_max_connections(1).await;
seed_fresh_layout(&pool).await;
sqlx::query("SET DateStyle TO 'SQL, DMY'")
.execute(&pool)
@@ -3658,13 +3866,12 @@ mod tests {
.expect("show TimeZone");
assert_eq!(date_style, "SQL, DMY");
assert_eq!(time_zone, "America/Los_Angeles");
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn audit_transaction_applies_and_restores_statement_timeout() {
let (pool, admin, schema) = scratch_pool_with_max_connections(1).await;
let (pool, _scratch) = scratch_pool_with_max_connections(1).await;
sqlx::query("SET statement_timeout TO '37s'")
.execute(&pool)
.await
@@ -3692,13 +3899,12 @@ mod tests {
.await
.expect("show restored statement timeout");
assert_eq!(restored, "37s");
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn statement_triggers_are_not_part_of_leaf_parity() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
sqlx::query(sqlx::AssertSqlSafe(format!(
@@ -3734,13 +3940,12 @@ mod tests {
&& table.missing_trigger_count() == 0
&& table.extra_trigger_count() == 0
}));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn canonical_name_with_wrong_bounds_is_a_real_creation_error() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
create_child(
@@ -3776,13 +3981,13 @@ mod tests {
.find(|table| table.table == "events")
.expect("events audit");
assert_eq!(events.months[0].kind, MonthCoverageKind::Uncovered);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn canonical_name_attached_to_wrong_parent_is_a_real_creation_error() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
let schema = scratch.name.as_str();
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
create_child(
@@ -3833,18 +4038,17 @@ mod tests {
)
"#,
)
.bind(&schema)
.bind(schema)
.fetch_one(&pool)
.await
.expect("check attachment");
assert!(!attached_to_events);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn unrelated_canonical_relation_is_a_real_creation_error() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
create_child(
@@ -3869,13 +4073,12 @@ mod tests {
assert!(relation_name_exists(&pool, "events_p2026_09")
.await
.expect("query unrelated relation"));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn anomalous_child_and_trigger_mismatch_degrade_without_aborting() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
create_child(
@@ -3923,7 +4126,6 @@ mod tests {
.iter()
.any(|child| child.extra_triggers == ["child_only_probe"])
}));
drop_schema(&admin, &schema).await;
}
}
}

83
crates/buzz-db/src/store/partition/maintenance_tests.rs
View File

@@ -328,8 +328,8 @@ mod postgres_tests {
use super::*;
use crate::store::partition::tests::postgres_tests::{
catalog_snapshot, create_child, drop_schema, fixed_now, scratch_pool,
scratch_pool_with_max_connections, seed_parents,
catalog_snapshot, create_child, fixed_now, scratch_pool, scratch_pool_with_max_connections,
seed_parents,
};
/// The repaired incident layout for both managed parents, optionally with
@@ -441,7 +441,7 @@ mod postgres_tests {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn empty_catch_all_advances_to_canonical_layout_and_is_idempotent() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, true).await;
let audit = maintain(&pool).await.expect("advance");
@@ -474,13 +474,12 @@ mod postgres_tests {
let before = catalog_snapshot(&pool).await;
maintain(&pool).await.expect("idempotent rerun");
assert_eq!(catalog_snapshot(&pool).await, before);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn canonical_catch_all_name_is_replaced_in_one_transaction() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
for table in PARTITIONED_TABLES {
create_child(
@@ -506,13 +505,12 @@ mod postgres_tests {
.children
.iter()
.any(|child| child.name == format!("{}_p2026_09", table.table))));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn populated_catch_all_is_refused_before_any_lock() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
sqlx::query("INSERT INTO events (created_at, alternate_at) VALUES ('2027-02-01', now())")
.execute(&pool)
@@ -544,13 +542,12 @@ mod postgres_tests {
.collect::<Vec<_>>()
};
assert_eq!(events_only(&after), events_only(&before));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn counterpart_contention_fails_fast_and_rolls_back() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, true).await;
let before = catalog_snapshot(&pool).await;
let holder = hold_lock(
@@ -575,13 +572,12 @@ mod postgres_tests {
assert_eq!(catalog_snapshot(&pool).await, before);
assert_advanced(&maintain(&pool).await.expect("advance after release"));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn counterpart_reader_does_not_block_advancement() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, true).await;
let holder = hold_lock(&pool, "SELECT count(*) FROM communities").await;
@@ -589,13 +585,12 @@ mod postgres_tests {
drop(holder);
assert_advanced(&result.expect("advance beside a counterpart reader"));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn writer_checking_a_foreign_key_behind_maintenance_does_not_deadlock() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
seed_repaired_layout(&pool, true).await;
// An in-flight insert: the parent first, then the foreign-key check.
let mut writer = hold_lock(&pool, "LOCK TABLE ONLY events IN ROW EXCLUSIVE MODE").await;
@@ -615,7 +610,7 @@ mod postgres_tests {
WHERE $1 = ANY(pg_blocking_pids(pid)))",
)
.bind(writer_pid)
.fetch_one(&admin)
.fetch_one(&scratch.admin)
.await
.expect("check queued maintenance");
if queued {
@@ -650,7 +645,6 @@ mod postgres_tests {
.expect("maintenance task")
.expect("advance after the writer commits"),
);
drop_schema(&admin, &schema).await;
}
async fn wait_until_blocked_by(admin: &PgPool, pid: i32, waiters: i64) {
@@ -673,7 +667,7 @@ mod postgres_tests {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn foreign_key_committed_while_waiting_on_the_parent_is_locked_nowait() {
let (pool, admin, schema) = scratch_pool_with_max_connections(6).await;
let (pool, scratch) = scratch_pool_with_max_connections(6).await;
seed_repaired_layout(&pool, true).await;
sqlx::query("CREATE TABLE reviewers (id UUID PRIMARY KEY)")
.execute(&pool)
@@ -708,12 +702,12 @@ mod postgres_tests {
.expect("roll back counterpart writer");
}
});
wait_until_blocked_by(&admin, migration_pid, 1).await;
wait_until_blocked_by(&scratch.admin, migration_pid, 1).await;
let maintenance = tokio::spawn({
let pool = pool.clone();
async move { maintain(&pool).await }
});
wait_until_blocked_by(&admin, migration_pid, 2).await;
wait_until_blocked_by(&scratch.admin, migration_pid, 2).await;
migration.commit().await.expect("commit migration");
let result = maintenance.await.expect("maintenance task");
@@ -733,7 +727,6 @@ mod postgres_tests {
release_writer.send(()).expect("release counterpart writer");
writer.await.expect("counterpart writer task");
assert_advanced(&maintain(&pool).await.expect("advance after release"));
drop_schema(&admin, &schema).await;
}
async fn backend_pid(connection: &mut PgConnection) -> i32 {
@@ -753,7 +746,7 @@ mod postgres_tests {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn foreign_key_referencing_the_parent_committed_while_waiting_is_refused() {
let (pool, admin, schema) = scratch_pool_with_max_connections(6).await;
let (pool, scratch) = scratch_pool_with_max_connections(6).await;
seed_repaired_layout(&pool, true).await;
sqlx::query(
"CREATE TABLE event_refs \
@@ -775,7 +768,7 @@ mod postgres_tests {
let pool = pool.clone();
async move { maintain(&pool).await }
});
wait_until_blocked_by(&admin, migration_pid, 1).await;
wait_until_blocked_by(&scratch.admin, migration_pid, 1).await;
migration.commit().await.expect("commit migration");
let result = maintenance.await.expect("maintenance task");
@@ -783,13 +776,12 @@ mod postgres_tests {
refused_events_only(&result, "events is referenced by a foreign key"),
"{result:?}"
);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn catch_all_foreign_keys_outside_the_parent_are_refused() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, true).await;
sqlx::query("CREATE TABLE reviewers (id UUID PRIMARY KEY)")
.execute(&pool)
@@ -842,13 +834,12 @@ mod postgres_tests {
}
drop(holder);
assert_advanced(&maintain(&pool).await.expect("advance without the keys"));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn catch_all_foreign_key_committed_while_waiting_on_the_parent_is_refused() {
let (pool, admin, schema) = scratch_pool_with_max_connections(6).await;
let (pool, scratch) = scratch_pool_with_max_connections(6).await;
seed_repaired_layout(&pool, true).await;
sqlx::query("CREATE TABLE reviewers (id UUID PRIMARY KEY)")
.execute(&pool)
@@ -868,7 +859,7 @@ mod postgres_tests {
let pool = pool.clone();
async move { maintain(&pool).await }
});
wait_until_blocked_by(&admin, reader_pid, 1).await;
wait_until_blocked_by(&scratch.admin, reader_pid, 1).await;
migration.commit().await.expect("commit migration");
reader.commit().await.expect("release parent");
@@ -885,13 +876,12 @@ mod postgres_tests {
.await
.expect("count catch-all key");
assert_eq!(kept, 1);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn deadlock_victim_is_classified_as_a_lock_timeout() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_parents(&pool).await;
let mut first = hold_lock(&pool, "LOCK TABLE ONLY events IN ACCESS EXCLUSIVE MODE").await;
let mut second = hold_lock(
@@ -925,13 +915,12 @@ mod postgres_tests {
PartitionMaintenanceOutcome::LockTimeout
);
drop((first, second));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn parent_contention_times_out_without_blocking_the_other_table() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, true).await;
let holder = hold_lock(&pool, "LOCK TABLE ONLY events IN ACCESS SHARE MODE").await;
@@ -961,13 +950,12 @@ mod postgres_tests {
audited_at: audit.audited_at,
tables: delivery_log,
});
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn held_maintenance_lock_skips_without_ddl() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
let before = catalog_snapshot(&pool).await;
let holder = hold_lock(
@@ -982,13 +970,12 @@ mod postgres_tests {
.expect("skipped_locked is not a failure");
assert_eq!(catalog_snapshot(&pool).await, before);
drop(holder);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn concurrent_runners_converge_on_one_layout() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, true).await;
let (first, second) = tokio::join!(maintain(&pool), maintain(&pool));
@@ -1012,13 +999,12 @@ mod postgres_tests {
}
// Whatever a fail-fast loser left undone, the next pass converges.
assert_advanced(&maintain(&pool).await.expect("follow-up pass"));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn client_deadline_rolls_back_and_discards_the_session() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
let before = catalog_snapshot(&pool).await;
// The parent lock wait (2s) outlasts the 300ms client deadline.
@@ -1050,7 +1036,7 @@ mod postgres_tests {
"SELECT pid FROM pg_stat_activity WHERE $1 = ANY(pg_blocking_pids(pid))",
)
.bind(holder_pid)
.fetch_all(&admin)
.fetch_all(&scratch.admin)
.await
.expect("sessions blocked by the holder");
let mut gone = blocked.is_empty();
@@ -1063,7 +1049,7 @@ mod postgres_tests {
"SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE pid = ANY($1))",
)
.bind(&blocked)
.fetch_one(&admin)
.fetch_one(&scratch.admin)
.await
.expect("check abandoned backend");
gone = !alive;
@@ -1086,13 +1072,12 @@ mod postgres_tests {
.collect::<Vec<_>>()
};
assert_eq!(events_only(&after), events_only(&before));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn catch_all_with_child_only_trigger_is_refused() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
sqlx::query(
"CREATE TRIGGER catch_all_only BEFORE INSERT ON events_p_future_next \
@@ -1120,13 +1105,12 @@ mod postgres_tests {
.collect::<Vec<_>>()
};
assert_eq!(events_only(&after), events_only(&before));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn stale_audit_after_a_concurrent_advance_is_a_noop() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
// A runner audits, then loses the race to a runner that commits first.
let stale = audit_partition_catalog_at(&pool, PARTITION_MANAGER_MONTHS_AHEAD, fixed_now())
@@ -1167,13 +1151,12 @@ mod postgres_tests {
.await
.expect("losing runner");
assert_advanced(&reported);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn replan_that_finds_only_disabled_work_is_skipped_disabled() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
sqlx::query("DROP TABLE events_p_future_next")
.execute(&pool)
@@ -1210,13 +1193,12 @@ mod postgres_tests {
(PartitionMaintenanceOutcome::SkippedDisabled, None)
);
assert_eq!(catalog_snapshot(&pool).await, before);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn audit_retries_when_a_listed_partition_is_dropped() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
let mut dropper = pool.begin().await.expect("begin dropper");
let dropper_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
@@ -1242,7 +1224,7 @@ mod postgres_tests {
WHERE $1 = ANY(pg_blocking_pids(pid)))",
)
.bind(dropper_pid)
.fetch_one(&admin)
.fetch_one(&scratch.admin)
.await
.expect("check audit queue");
if queued {
@@ -1263,13 +1245,12 @@ mod postgres_tests {
.children
.iter()
.all(|child| child.name != "events_p_future_next"));
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn unrelated_catch_all_name_is_an_operator_error() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
sqlx::query("CREATE TABLE events_p_future (unrelated BOOLEAN)")
.execute(&pool)
@@ -1297,13 +1278,12 @@ mod postgres_tests {
.filter(|(name, ..)| name.starts_with("events_"))
.count()
);
drop_schema(&admin, &schema).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn incoming_foreign_key_is_refused_under_lock() {
let (pool, admin, schema) = scratch_pool().await;
let (pool, _scratch) = scratch_pool().await;
seed_repaired_layout(&pool, false).await;
sqlx::query(
"CREATE TABLE event_refs (created_at TIMESTAMPTZ, alternate_at TIMESTAMPTZ, id BIGINT, \
@@ -1325,6 +1305,5 @@ mod postgres_tests {
.iter()
.any(|(name, ..)| name == "events_p_future_next"));
assert_ne!(after, before, "delivery_log still advances");
drop_schema(&admin, &schema).await;
}
}