From 52d13efc08ec2b80878f24672913ac465b37e148 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Wed, 14 Oct 2026 08:27:26 -0400 Subject: [PATCH] fix(buzz-db): delete workflow and membership children explicitly (#8212) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What Workflow deletion and relay-member removal relied on `ON DELETE CASCADE` to remove child rows. Both paths now delete their own children inside the same transaction, so a later migration can drop those foreign-key actions without changing behavior. **Workflow deletion** (`store/workflow/deletion.rs`, the coordinate path used by NIP-09 deletes of workflow definitions): - Locks the target workflow row with `SELECT … FOR UPDATE`, under the same owner and coordinate predicate as before. A concurrent run or fire insert has to lock that row for its foreign key, so the child deletes see every committed child. - Deletes scheduled fires first. A new fire has to lock the workflow row, and a cron attach locks the fire before the run, so deleting fires before the run lock keeps the lock order consistent and avoids a deadlock. - Then locks the workflow's runs `FOR UPDATE` in `id` order. An approval that names another workflow but references one of these runs key-share locks only the run, so the workflow lock alone doesn't exclude it. - `delete_workflow_with_children` then deletes approvals (for the workflow or any of its runs), then runs, then the workflow, and returns its `channel_id`. **Relay-member removal** (`store/relay_members.rs`): `remove_relay_member` and `remove_relay_member_if_role` share a new `delete_member_with_acceptances` helper. It locks the member row under the existing predicate (any non-owner, or an exact role), deletes `join_policy_acceptances`, then deletes the member, in one transaction. The `RemoveResult` outcomes are unchanged. **Removed:** `delete_workflow` (no callers) and `delete_workflow_for_owner` (test-only), plus their `Db` wrappers. The community-confinement test now drives the production coordinate deletion. ## Tests - `test_support::retire_foreign_key_delete_actions` rewrites every `CASCADE`, `SET NULL` and `SET DEFAULT` foreign key in the test's own per-test database as `NO ACTION`, which is the target schema of the follow-up migration. It asserts that none remain. All triggers stay live, including foreign-key checks and the community write fence. It refuses to run unless `current_database()` has the per-test wrapper's exact shape (`buzz_nt_` plus 24 hex digits), which excludes the run's `_desired` template. It checks the name only. - `workflow_deletion_removes_children_without_cascades` and `member_removal_deletes_acceptances_without_cascades` run the production paths on that schema and require every child row to be gone. One approval references another workflow but this workflow's run, which pins the `run_id` branch. The owner, role-mismatch and not-found outcomes still hold. - `workflow_deletion_waits_for_and_removes_concurrent_run` and `member_removal_waits_for_and_removes_concurrent_acceptance` hold a child insert open in a second transaction, wait until the removal is blocked on a lock, and then commit. The removal must succeed with no children left. - `workflow_deletion_waits_for_concurrent_fire_attach_without_deadlock` holds a fire row the way the cron attach does, then attaches a run to it while the deletion waits. Both must succeed. - `workflow_deletion_waits_for_and_removes_concurrent_cross_workflow_approval` does the same with an approval for another workflow that references this workflow's run. - The bridge test schema now includes `join_policy_acceptances`. ## What got simpler Two delete paths no longer depend on implicit constraint actions, and two unused workflow-delete functions are gone. This is the first step of retiring the `CASCADE` and `SET NULL` foreign keys. **Identified debt (later PRs):** `channel_members` and `personal_read_frontiers` (no hard-delete path found yet for their parents), the deletion `manifest_keys` cascade, the `SET NULL` actions on `executor_heartbeats` and `product_feedback`, and the `users` self-reference. That last one is a composite `(community_id, agent_owner_pubkey)` `ON DELETE SET NULL` that would try to null the `NOT NULL` `community_id`. Also, a scheduled fire for workflow B can link to a run of A (that FK is already `NO ACTION`), which would block A's run delete; the scheduler always attaches a run to its own workflow's fire, so this needs same-workflow enforcement in the cascade work. ## Verification Head `081be4c0d`, tree `ab649487`, on a Linux cloud workstation: fmt, buzz-db and workspace clippy, source policy and buzz-db lib pass; the six new tests pass; buzz-db PostgreSQL passes 490/490 and buzz-relay PostgreSQL 378/378. A sixth mutation, moving the fire delete back after the run lock, makes `workflow_deletion_waits_for_concurrent_fire_attach_without_deadlock` fail with `40P01` (deadlock detected). Previous head: Head `787c7c42d`, tree `6b9cf5d5`, on a Linux cloud workstation: fmt, buzz-db and workspace clippy, source policy and buzz-db lib pass; the new tests pass 5/5; buzz-relay PostgreSQL passes 378/378; buzz-db PostgreSQL passes 488/489, the one failure being `parent_contention_times_out_without_blocking_the_other_table` in `store/partition/`, which this PR doesn't touch and which passes on its own at the same tree. A fifth mutation, removing the run lock, makes the new cross-workflow approval test fail with the reviewer's exact 23503 on `workflow_approvals_community_id_run_id_fkey`. Previous head `00cba2f4a`, tree `2f89fed4`, base `e9269cbdf`. Results from a Linux cloud workstation: - fmt, buzz-db and workspace clippy, the source policy and buzz-db lib all pass. - The buzz-db PostgreSQL suite passes 488/488. - The buzz-relay PostgreSQL suite passes 378/378. Each of four mutations makes exactly one of the new tests fail with the foreign-key violation you'd expect (23503): - swapping the fires and runs deletes - deleting approvals by `workflow_id` only - removing the workflow row lock - removing the member row lock GitHub CI is green at this head. buzz-review-completed --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/store/community.rs | 38 +-- crates/buzz-db/src/store/relay_members.rs | 199 +++++++++++-- crates/buzz-db/src/store/workflow.rs | 82 +---- crates/buzz-db/src/store/workflow/deletion.rs | 74 ++++- .../src/store/workflow/deletion_tests.rs | 279 +++++++++++++++++- crates/buzz-db/src/test_support.rs | 90 ++++++ crates/buzz-relay/src/api/bridge.rs | 4 +- 7 files changed, 619 insertions(+), 147 deletions(-) diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index 6cf43ee7c5..2407535958 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -1180,39 +1180,7 @@ mod postgres_tests { } async fn named_contender_db(application_name: &str) -> Db { - use std::str::FromStr as _; - let options = - sqlx::postgres::PgConnectOptions::from_str(&crate::test_support::database_url()) - .expect("parse test database URL") - .application_name(application_name); - let pool = sqlx::postgres::PgPoolOptions::new() - .max_connections(1) - .connect_with(options) - .await - .expect("connect contender DB"); - Db::from_pool(pool) - } - - async fn wait_for_lock_wait(db: &Db, application_name: &str) { - tokio::time::timeout(std::time::Duration::from_secs(5), async { - loop { - let waiting: bool = sqlx::query_scalar( - "SELECT EXISTS(SELECT 1 FROM pg_stat_activity \ - WHERE datname = current_database() AND application_name = $1 \ - AND wait_event_type = 'Lock')", - ) - .bind(application_name) - .fetch_one(&db.pool) - .await - .expect("inspect contender lock wait"); - if waiting { - break; - } - tokio::task::yield_now().await; - } - }) - .await - .expect("contender reached a lock wait"); + Db::from_pool(crate::test_support::named_pool(application_name).await) } /// An archive that queues behind an ownership transfer must re-check @@ -1256,7 +1224,7 @@ mod postgres_tests { .await } }); - wait_for_lock_wait(&db, &transfer_name).await; + crate::test_support::wait_for_lock_wait(&db.pool, &transfer_name).await; let archive_name = format!("archive-transfer-archive-{}", Uuid::new_v4().simple()); let archive_db = named_contender_db(&archive_name).await; @@ -1269,7 +1237,7 @@ mod postgres_tests { .await } }); - wait_for_lock_wait(&db, &archive_name).await; + crate::test_support::wait_for_lock_wait(&db.pool, &archive_name).await; owner_rows_gate .rollback() diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 8db2082014..742e9173f2 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -243,11 +243,51 @@ pub enum RemoveResult { RoleMismatch, } +/// Deletes a membership and its join-policy acceptances in one transaction. +/// +/// With `expected_role`, only a member holding exactly that role is removed; +/// without it, any non-owner is. The member row is locked under that predicate +/// first, so a concurrent acceptance insert (which must lock the member row for +/// its foreign key) either commits before the child delete sees it or fails. +/// Children go before the parent so no foreign-key action is needed. +async fn delete_member_with_acceptances( + connection: &mut sqlx::PgConnection, + community: CommunityId, + pubkey: &str, + expected_role: Option<&str>, +) -> Result { + let mut tx = sqlx::Connection::begin(connection).await?; + let locked = sqlx::query( + "SELECT 1 FROM relay_members WHERE community_id = $1 AND pubkey = $2 \ + AND CASE WHEN $3::text IS NULL THEN role <> 'owner' ELSE role = $3 END \ + FOR UPDATE", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .bind(expected_role) + .fetch_optional(&mut *tx) + .await?; + if locked.is_none() { + return Ok(false); + } + sqlx::query("DELETE FROM join_policy_acceptances WHERE community_id = $1 AND pubkey = $2") + .bind(community.as_uuid()) + .bind(pubkey) + .execute(&mut *tx) + .await?; + sqlx::query("DELETE FROM relay_members WHERE community_id = $1 AND pubkey = $2") + .bind(community.as_uuid()) + .bind(pubkey) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(true) +} + /// Removes a relay member atomically, refusing to delete the owner. /// -/// Uses a single conditional `DELETE … WHERE role <> 'owner'` so the -/// owner-protection check and the deletion are one atomic operation — -/// no TOCTOU race between a separate read and delete. +/// The owner-protection check and the deletion run in one transaction under a +/// row lock, so there is no TOCTOU race between a separate read and delete. pub async fn remove_relay_member( pool: &PgPool, community: CommunityId, @@ -255,20 +295,11 @@ pub async fn remove_relay_member( ) -> Result { let mut connection = observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; - let result = sqlx::query( - "DELETE FROM relay_members \ - WHERE community_id = $1 AND pubkey = $2 AND role <> 'owner'", - ) - .bind(community.as_uuid()) - .bind(pubkey) - .execute(&mut *connection) - .await?; - - if result.rows_affected() > 0 { + if delete_member_with_acceptances(&mut connection, community, pubkey, None).await? { return Ok(RemoveResult::Removed); } - // rows_affected == 0: either not found or is owner. One cheap read to + // Nothing was locked: either not found or is owner. One cheap read to // distinguish the two cases so callers can return the right error message. let exists = sqlx::query("SELECT 1 FROM relay_members WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) @@ -285,9 +316,8 @@ pub async fn remove_relay_member( /// Removes a relay member only if their current role matches `expected_role`. /// -/// The delete and the role check are collapsed into a single -/// `DELETE … WHERE pubkey = $1 AND role = $2`, making the operation atomic — -/// no TOCTOU race between a prior read and this delete. +/// The role check and the delete run in one transaction under a row lock, so +/// there is no TOCTOU race between a prior read and this delete. /// /// Returns: /// - `Removed` — row was deleted. @@ -304,20 +334,13 @@ pub async fn remove_relay_member_if_role( ) -> Result { let mut connection = observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; - let result = sqlx::query( - "DELETE FROM relay_members WHERE community_id = $1 AND pubkey = $2 AND role = $3", - ) - .bind(community.as_uuid()) - .bind(pubkey) - .bind(expected_role) - .execute(&mut *connection) - .await?; - - if result.rows_affected() > 0 { + if delete_member_with_acceptances(&mut connection, community, pubkey, Some(expected_role)) + .await? + { return Ok(RemoveResult::Removed); } - // rows_affected == 0: either not found or role changed. One cheap read to + // Nothing was locked: either not found or role changed. One cheap read to // distinguish the cases so callers can return the right error message. let row = sqlx::query("SELECT role FROM relay_members WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) @@ -1392,6 +1415,126 @@ mod postgres_tests { ); } + async fn acceptances(pool: &PgPool, community: CommunityId, pubkey: &str) -> i64 { + sqlx::query_scalar( + "SELECT count(*) FROM join_policy_acceptances WHERE community_id = $1 AND pubkey = $2", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .fetch_one(pool) + .await + .expect("count acceptances") + } + + /// Both removal paths delete acceptance evidence themselves: with every + /// foreign-key delete action retired, nothing cascades. Owners and a role + /// mismatch keep theirs. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn member_removal_deletes_acceptances_without_cascades() { + let pool = setup_pool().await; + crate::test_support::retire_foreign_key_delete_actions(&pool).await; + let (community, owner) = owned_community(&pool).await; + let version = "a".repeat(64); + sqlx::query( + "INSERT INTO join_policy_acceptances (community_id, pubkey, policy_version) \ + VALUES ($1, $2, $3)", + ) + .bind(community.as_uuid()) + .bind(&owner) + .bind(&version) + .execute(&pool) + .await + .expect("owner acceptance"); + let member = test_pubkey(); + let admin = test_pubkey(); + for (pubkey, role) in [(&member, "member"), (&admin, "admin")] { + claim_relay_membership(&pool, community, pubkey, role, Some(&version)) + .await + .expect("claim membership"); + assert_eq!(acceptances(&pool, community, pubkey).await, 1); + } + + assert_eq!( + remove_relay_member(&pool, community, &owner) + .await + .expect("owner removal"), + RemoveResult::IsOwner + ); + assert_eq!(acceptances(&pool, community, &owner).await, 1); + assert_eq!( + remove_relay_member_if_role(&pool, community, &admin, "member") + .await + .expect("mismatched removal"), + RemoveResult::RoleMismatch + ); + assert_eq!(acceptances(&pool, community, &admin).await, 1); + + assert_eq!( + remove_relay_member(&pool, community, &member) + .await + .expect("member removal"), + RemoveResult::Removed + ); + assert_eq!(acceptances(&pool, community, &member).await, 0); + assert_eq!( + remove_relay_member_if_role(&pool, community, &admin, "admin") + .await + .expect("role removal"), + RemoveResult::Removed + ); + assert_eq!(acceptances(&pool, community, &admin).await, 0); + } + + /// An acceptance inserted concurrently with the removal holds the member + /// row's key-share lock. Removal must wait for it at its row lock and then + /// delete it; with no cascade, a removal that took its child snapshot first + /// would fail the foreign-key check on the member. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn member_removal_waits_for_and_removes_concurrent_acceptance() { + let pool = setup_pool().await; + crate::test_support::retire_foreign_key_delete_actions(&pool).await; + let (community, _owner) = owned_community(&pool).await; + let member = test_pubkey(); + claim_relay_membership(&pool, community, &member, "member", Some(&"a".repeat(64))) + .await + .expect("claim membership"); + + let mut inserter = pool.begin().await.expect("begin acceptance insert"); + sqlx::query( + "INSERT INTO join_policy_acceptances (community_id, pubkey, policy_version) \ + VALUES ($1, $2, $3)", + ) + .bind(community.as_uuid()) + .bind(&member) + .bind("b".repeat(64)) + .execute(&mut *inserter) + .await + .expect("insert concurrent acceptance"); + + let name = format!("member-remove-{}", Uuid::new_v4().simple()); + let remover = crate::test_support::named_pool(&name).await; + let removal = tokio::spawn({ + let member = member.clone(); + async move { remove_relay_member(&remover, community, &member).await } + }); + crate::test_support::wait_for_lock_wait(&pool, &name).await; + inserter + .commit() + .await + .expect("commit concurrent acceptance"); + + assert_eq!( + removal + .await + .expect("join removal") + .expect("removal after concurrent acceptance"), + RemoveResult::Removed + ); + assert_eq!(acceptances(&pool, community, &member).await, 0); + } + async fn owned_community(pool: &PgPool) -> (CommunityId, String) { let community = make_test_community(pool).await; let owner = test_pubkey(); diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index da02c7f75d..c89bf0cbf0 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -775,56 +775,6 @@ pub async fn disable_workflows_for_owner_in_channel_on_conn( Ok(affected) } -/// Delete a workflow and all its runs/approvals (CASCADE). -/// -/// NOTE: see the cache-invalidation note on [`update_workflow`]. The relay's -/// deletion path uses [`delete_workflow_for_owner`], which returns the -/// `channel_id` needed for invalidation. (No current callers.) -pub async fn delete_workflow(pool: &PgPool, community_id: CommunityId, id: Uuid) -> Result<()> { - let affected = sqlx::query("DELETE FROM workflows WHERE community_id = $1 AND id = $2") - .bind(community_id.as_uuid()) - .bind(id) - .execute(pool) - .await? - .rows_affected(); - - if affected == 0 { - return Err(DbError::NotFound(format!("workflow {id}"))); - } - Ok(()) -} - -/// Delete a workflow only when it belongs to `owner_pubkey`. -/// -/// Used by event-driven deletion paths where the workflow UUID is attacker -/// controlled. Keeping the owner predicate in the DELETE statement avoids a -/// check-then-delete race and ensures a caller cannot delete another user's -/// workflow just by learning its UUID. -/// -/// Returns the deleted workflow's `channel_id` so the caller can invalidate -/// the per-channel trigger cache without a separate lookup. -pub async fn delete_workflow_for_owner( - pool: &PgPool, - community_id: CommunityId, - id: Uuid, - owner_pubkey: &[u8], -) -> Result> { - let row = sqlx::query( - "DELETE FROM workflows WHERE community_id = $1 AND id = $2 AND owner_pubkey = $3 \ - RETURNING channel_id", - ) - .bind(community_id.as_uuid()) - .bind(id) - .bind(owner_pubkey) - .fetch_optional(pool) - .await?; - - match row { - Some(row) => Ok(row.try_get("channel_id")?), - None => Err(DbError::NotFound(format!("workflow {id}"))), - } -} - // -- Workflow Run CRUD -------------------------------------------------------- /// Insert a new workflow run. Returns the new run's UUID. @@ -1684,24 +1634,6 @@ impl Db { .await } - /// Delete a workflow and all its runs/approvals. - #[datastore_span(name = "delete_workflow", system = "postgresql")] - pub async fn delete_workflow(&self, community_id: CommunityId, id: Uuid) -> Result<()> { - crate::workflow::delete_workflow(&self.pool, community_id, id).await - } - - /// Delete a workflow only when it belongs to the provided owner. - /// Returns the deleted workflow's `channel_id`. - #[datastore_span(name = "delete_workflow_for_owner", system = "postgresql")] - pub async fn delete_workflow_for_owner( - &self, - community_id: CommunityId, - id: Uuid, - owner_pubkey: &[u8], - ) -> Result> { - crate::workflow::delete_workflow_for_owner(&self.pool, community_id, id, owner_pubkey).await - } - /// Find a workflow by owner pubkey and name within a community. Used for /// NIP-09 a-tag deletion where the d-tag is the workflow name (not UUID). #[datastore_span(name = "find_workflow_by_owner_and_name", system = "postgresql")] @@ -2656,9 +2588,9 @@ mod postgres_tests { } /// Issue 4 (workflow lifecycle): deleting `A/id` must not delete `B/id` - /// when both communities hold the same workflow UUID. Pre-fix - /// `delete_workflow` predicated only on `id`, so a NIP-09 a-tag deletion in - /// one community would erase the colliding workflow in every community. + /// when both communities hold the same workflow UUID. Pre-fix the workflow + /// delete predicated only on `id`, so a NIP-09 a-tag deletion in one + /// community would erase the colliding workflow in every community. #[tokio::test] #[ignore = "requires Postgres"] async fn workflow_delete_is_confined_to_its_community() { @@ -2684,7 +2616,13 @@ mod postgres_tests { ) .await; - delete_workflow(&pool, community_a, shared_workflow_id) + crate::Db::from_pool(pool.clone()) + .delete_workflow_by_coordinate( + community_a, + &[0xb2; 32], + &shared_workflow_id.to_string(), + chrono::Utc::now().timestamp(), + ) .await .expect("delete A's workflow"); diff --git a/crates/buzz-db/src/store/workflow/deletion.rs b/crates/buzz-db/src/store/workflow/deletion.rs index 27fde32b47..84f4b82845 100644 --- a/crates/buzz-db/src/store/workflow/deletion.rs +++ b/crates/buzz-db/src/store/workflow/deletion.rs @@ -4,7 +4,6 @@ use buzz_core::{kind::KIND_WORKFLOW_DEF, CommunityId, StoredEvent}; use buzz_datastore_tracing::datastore_span; use chrono::{DateTime, Utc}; use nostr::Event; -use sqlx::Row; use uuid::Uuid; use crate::AdmittedTx; @@ -115,13 +114,16 @@ async fn delete_workflow_in_transaction( } // UUID coordinates are canonical; retain the legacy name-based path. - // The owner predicate remains in the mutation, not just a prior check. + // The owner predicate is in the locking SELECT; FOR UPDATE keeps it + // binding for the deletes below. let workflow_id = Uuid::parse_str(d_tag).ok(); - let row = sqlx::query( - "DELETE FROM workflows WHERE community_id = $1 AND owner_pubkey = $2 \ + // Lock the workflow first: a concurrent run or fire insert must lock it for + // its foreign key, so the child deletes below see every committed child. + let target: Option = sqlx::query_scalar( + "SELECT id FROM workflows WHERE community_id = $1 AND owner_pubkey = $2 \ AND id = COALESCE($3::uuid, (SELECT id FROM workflows \ WHERE community_id = $1 AND owner_pubkey = $2 AND name = $4 LIMIT 1)) \ - RETURNING channel_id", + FOR UPDATE", ) .bind(community_id.as_uuid()) .bind(owner_pubkey) @@ -129,6 +131,10 @@ async fn delete_workflow_in_transaction( .bind(d_tag) .fetch_optional(tx.conn()) .await?; + let deleted_channel = match target { + Some(id) => Some(delete_workflow_with_children(tx, id).await?), + None => None, + }; let definitions = sqlx::query( "UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND kind = $2 \ AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL AND created_at <= $5", @@ -140,17 +146,65 @@ async fn delete_workflow_in_transaction( .bind(cutoff) .execute(tx.conn()) .await?; - let changed = row.is_some() || definitions.rows_affected() > 0; - let channel_id = row - .map(|row| row.try_get("channel_id")) - .transpose()? - .flatten(); + let changed = deleted_channel.is_some() || definitions.rows_affected() > 0; + let channel_id = deleted_channel.flatten(); Ok(WorkflowDeletionOutcome { changed, channel_id, }) } +/// Delete a locked workflow and the rows that hang off it, children first, so +/// no foreign-key action is needed: scheduled fires, approvals (of the workflow +/// or any of its runs), then runs. Returns the workflow's `channel_id`. +async fn delete_workflow_with_children(tx: &mut AdmittedTx, id: Uuid) -> Result> { + let community = *tx.community().as_uuid(); + // Fires go before the run lock. A new fire must key-share lock the workflow, + // which is already locked. Attaching a run to a fire locks the fire row and + // then the run, so taking the run lock first would invert that order and can + // deadlock; deleting the fire first waits for any in-flight attach instead. + sqlx::query( + "DELETE FROM scheduled_workflow_fires WHERE community_id = $1 AND workflow_id = $2", + ) + .bind(community) + .bind(id) + .execute(tx.conn()) + .await?; + // The workflow lock stops new runs, but an approval naming another workflow + // can still reference one of these runs; it key-share locks only that run. + // Lock the runs (in a fixed order) so such an insert either commits before + // the approval delete below or waits until this transaction ends. + sqlx::query( + "SELECT id FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 \ + ORDER BY id FOR UPDATE", + ) + .bind(community) + .bind(id) + .execute(tx.conn()) + .await?; + sqlx::query( + "DELETE FROM workflow_approvals WHERE community_id = $1 AND (workflow_id = $2 \ + OR run_id IN (SELECT id FROM workflow_runs \ + WHERE community_id = $1 AND workflow_id = $2))", + ) + .bind(community) + .bind(id) + .execute(tx.conn()) + .await?; + sqlx::query("DELETE FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2") + .bind(community) + .bind(id) + .execute(tx.conn()) + .await?; + Ok(sqlx::query_scalar( + "DELETE FROM workflows WHERE community_id = $1 AND id = $2 RETURNING channel_id", + ) + .bind(community) + .bind(id) + .fetch_one(tx.conn()) + .await?) +} + #[cfg(test)] #[path = "deletion_tests.rs"] mod postgres_tests; diff --git a/crates/buzz-db/src/store/workflow/deletion_tests.rs b/crates/buzz-db/src/store/workflow/deletion_tests.rs index 0dd4909179..7210d2efb7 100644 --- a/crates/buzz-db/src/store/workflow/deletion_tests.rs +++ b/crates/buzz-db/src/store/workflow/deletion_tests.rs @@ -174,7 +174,10 @@ async fn workflow_deletion_preserves_newer_definition_and_cleans_legacy_orphan() assert_present(&db, &query, id).await; // Old relay releases deleted the executable row, but not the definition. - db.delete_workflow_for_owner(community, id, &owner) + sqlx::query("DELETE FROM workflows WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(id) + .execute(&db.pool) .await .expect("legacy deletion"); assert_eq!( @@ -360,3 +363,277 @@ async fn quiescing_community_rejects_workflow_deletion_at_admission_before_repla .expect("count deletion request"); assert_eq!(stored, 0, "a rejected deletion request must not be stored"); } + +async fn count_where(db: &Db, table: &str, column: &str, community: CommunityId, id: Uuid) -> i64 { + sqlx::query_scalar(sqlx::AssertSqlSafe(format!( + "SELECT count(*) FROM {table} WHERE community_id = $1 AND {column} = $2" + ))) + .bind(community.as_uuid()) + .bind(id) + .fetch_one(&db.pool) + .await + .expect("count children") +} + +/// With every foreign-key delete action retired, the coordinate delete must +/// remove approvals (of the workflow or of its runs), fires and runs itself, +/// in an order the remaining constraints accept. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_deletion_removes_children_without_cascades() { + let (db, community) = setup().await; + crate::test_support::retire_foreign_key_delete_actions(&db.pool).await; + let keys = Keys::generate(); + let owner = keys.public_key().to_bytes(); + let id = Uuid::new_v4(); + let d_tag = id.to_string(); + let now = Timestamp::now().as_secs(); + let query = seed(&db, community, &keys, id, &d_tag, now).await; + let other = Uuid::new_v4(); + seed(&db, community, &keys, other, &other.to_string(), now).await; + let run_id = create_workflow_run(&db.pool, community, id, None, None) + .await + .expect("run"); + sqlx::query( + "INSERT INTO scheduled_workflow_fires \ + (community_id, workflow_id, scheduled_for, workflow_run_id) VALUES ($1, $2, NOW(), $3)", + ) + .bind(community.as_uuid()) + .bind(id) + .bind(run_id) + .execute(&db.pool) + .await + .expect("scheduled claim linked to run"); + // One approval hangs off the workflow; the other names a different + // workflow but this workflow's run, so only the run_id branch reaches it. + for (token, workflow_id) in [("workflow-approval", id), ("run-approval", other)] { + crate::workflow::create_approval( + &db.pool, + crate::workflow::CreateApprovalParams { + community_id: community, + token, + workflow_id, + run_id, + step_id: "gate", + step_index: 0, + approver_spec: "@anyone", + expires_at: chrono::Utc::now() + chrono::Duration::hours(1), + }, + ) + .await + .expect("approval"); + } + + let outcome = db + .delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64) + .await + .expect("delete without cascades"); + assert!(outcome.changed); + assert_absent(&db, &query, id).await; + for (table, column, key) in [ + ("workflow_approvals", "workflow_id", id), + ("workflow_approvals", "run_id", run_id), + ("scheduled_workflow_fires", "workflow_id", id), + ("workflow_runs", "workflow_id", id), + ] { + assert_eq!( + count_where(&db, table, column, community, key).await, + 0, + "{table} rows by {column} must be deleted explicitly" + ); + } + assert!(get_workflow(&db.pool, community, other).await.is_ok()); +} + +/// A run inserted concurrently with the delete holds the workflow's key-share +/// lock. The delete must wait for it at its row lock and then remove the run; +/// with no cascade, a delete that took its child snapshot first would fail the +/// foreign-key check on the workflow. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_deletion_waits_for_and_removes_concurrent_run() { + let (db, community) = setup().await; + crate::test_support::retire_foreign_key_delete_actions(&db.pool).await; + let keys = Keys::generate(); + let owner = keys.public_key().to_bytes(); + let id = Uuid::new_v4(); + let d_tag = id.to_string(); + let now = Timestamp::now().as_secs(); + seed(&db, community, &keys, id, &d_tag, now).await; + + let mut inserter = db.pool.begin().await.expect("begin run insert"); + sqlx::query( + "INSERT INTO workflow_runs (community_id, id, workflow_id, status, current_step, execution_trace) \ + VALUES ($1, $2, $3, 'pending', 0, '[]')", + ) + .bind(community.as_uuid()) + .bind(Uuid::new_v4()) + .bind(id) + .execute(&mut *inserter) + .await + .expect("insert concurrent run"); + + let name = format!("workflow-delete-{}", Uuid::new_v4().simple()); + let deleter = Db::from_pool(crate::test_support::named_pool(&name).await); + let deletion = tokio::spawn(async move { + deleter + .delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64) + .await + }); + crate::test_support::wait_for_lock_wait(&db.pool, &name).await; + inserter.commit().await.expect("commit concurrent run"); + + let outcome = deletion + .await + .expect("join deletion") + .expect("delete after concurrent run"); + assert!(outcome.changed); + assert_eq!( + count_where(&db, "workflow_runs", "workflow_id", community, id).await, + 0 + ); +} + +/// An approval naming another workflow but one of this workflow's runs only +/// key-share locks that run, so the workflow lock does not exclude it. The +/// delete must lock the runs, wait for the insert, and then remove it; with no +/// cascade, committing it between the approval and run deletes would fail the +/// run delete's foreign-key check. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_deletion_waits_for_and_removes_concurrent_cross_workflow_approval() { + let (db, community) = setup().await; + crate::test_support::retire_foreign_key_delete_actions(&db.pool).await; + let keys = Keys::generate(); + let owner = keys.public_key().to_bytes(); + let id = Uuid::new_v4(); + let d_tag = id.to_string(); + let now = Timestamp::now().as_secs(); + seed(&db, community, &keys, id, &d_tag, now).await; + let other = Uuid::new_v4(); + seed(&db, community, &keys, other, &other.to_string(), now).await; + let run_id = create_workflow_run(&db.pool, community, id, None, None) + .await + .expect("run"); + + let mut inserter = db.pool.begin().await.expect("begin approval insert"); + sqlx::query( + "INSERT INTO workflow_approvals (community_id, token, workflow_id, run_id, step_id, \ + step_index, approver_spec, status, expires_at) \ + VALUES ($1, $2, $3, $4, 'gate', 0, '@anyone', 'pending', NOW() + INTERVAL '1 hour')", + ) + .bind(community.as_uuid()) + .bind(Uuid::new_v4().as_bytes().to_vec()) + .bind(other) + .bind(run_id) + .execute(&mut *inserter) + .await + .expect("insert concurrent cross-workflow approval"); + + let name = format!("workflow-delete-{}", Uuid::new_v4().simple()); + let deleter = Db::from_pool(crate::test_support::named_pool(&name).await); + let deletion = tokio::spawn(async move { + deleter + .delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64) + .await + }); + crate::test_support::wait_for_lock_wait(&db.pool, &name).await; + inserter.commit().await.expect("commit concurrent approval"); + + let outcome = deletion + .await + .expect("join deletion") + .expect("delete after concurrent approval"); + assert!(outcome.changed); + for (table, column, key) in [ + ("workflow_approvals", "run_id", run_id), + ("workflow_runs", "workflow_id", id), + ] { + assert_eq!( + count_where(&db, table, column, community, key).await, + 0, + "{table}" + ); + } + assert!(get_workflow(&db.pool, community, other).await.is_ok()); +} + +/// Attaching a run to a scheduled fire locks the fire row, then key-share +/// locks the run for its foreign key. The delete must take those two in the +/// same order (fire before run), so an attach in flight when the delete +/// starts completes and is then removed, rather than deadlocking with it. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_deletion_waits_for_concurrent_fire_attach_without_deadlock() { + let (db, community) = setup().await; + crate::test_support::retire_foreign_key_delete_actions(&db.pool).await; + let keys = Keys::generate(); + let owner = keys.public_key().to_bytes(); + let id = Uuid::new_v4(); + let d_tag = id.to_string(); + let now = Timestamp::now().as_secs(); + seed(&db, community, &keys, id, &d_tag, now).await; + let run_id = create_workflow_run(&db.pool, community, id, None, None) + .await + .expect("run"); + let scheduled_for: chrono::DateTime = sqlx::query_scalar( + "INSERT INTO scheduled_workflow_fires (community_id, workflow_id, scheduled_for) \ + VALUES ($1, $2, date_trunc('second', NOW())) RETURNING scheduled_for", + ) + .bind(community.as_uuid()) + .bind(id) + .fetch_one(&db.pool) + .await + .expect("unlinked scheduled claim"); + + // Hold the fire row as an attach does before its foreign-key check runs. + let mut attacher = db.pool.begin().await.expect("begin fire attach"); + sqlx::query( + "SELECT 1 FROM scheduled_workflow_fires \ + WHERE community_id = $1 AND workflow_id = $2 AND scheduled_for = $3 FOR UPDATE", + ) + .bind(community.as_uuid()) + .bind(id) + .bind(scheduled_for) + .execute(&mut *attacher) + .await + .expect("lock fire row"); + + let name = format!("workflow-delete-{}", Uuid::new_v4().simple()); + let deleter = Db::from_pool(crate::test_support::named_pool(&name).await); + let deletion = tokio::spawn(async move { + deleter + .delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64) + .await + }); + crate::test_support::wait_for_lock_wait(&db.pool, &name).await; + let attached = sqlx::query( + "UPDATE scheduled_workflow_fires SET workflow_run_id = $4 \ + WHERE community_id = $1 AND workflow_id = $2 AND scheduled_for = $3", + ) + .bind(community.as_uuid()) + .bind(id) + .bind(scheduled_for) + .bind(run_id) + .execute(&mut *attacher) + .await + .expect("attach run to fire without deadlock"); + assert_eq!(attached.rows_affected(), 1); + attacher.commit().await.expect("commit fire attach"); + + let outcome = deletion + .await + .expect("join deletion") + .expect("delete after concurrent fire attach"); + assert!(outcome.changed); + for (table, column, key) in [ + ("scheduled_workflow_fires", "workflow_id", id), + ("workflow_runs", "workflow_id", id), + ] { + assert_eq!( + count_where(&db, table, column, community, key).await, + 0, + "{table}" + ); + } +} diff --git a/crates/buzz-db/src/test_support.rs b/crates/buzz-db/src/test_support.rs index 73fa14c366..e94b8b1449 100644 --- a/crates/buzz-db/src/test_support.rs +++ b/crates/buzz-db/src/test_support.rs @@ -82,3 +82,93 @@ pub(crate) fn desired_state_schema_sql() -> String { pub(crate) fn desired_state_schema_dir() -> std::path::PathBuf { std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../schema") } + +/// Rewrite every `ON DELETE CASCADE`, `SET NULL` and `SET DEFAULT` foreign key +/// in this test's database as `NO ACTION`: the schema the cascade retirement +/// targets. Every trigger stays live, including the foreign-key checks and the +/// community fence, so a delete path exercised afterwards must remove its +/// children itself, in an order the remaining constraints accept. +/// +/// Each PostgreSQL test process owns its database (`crates/buzz-db/TESTING.md`), +/// so the rewrite does not reach other tests. It refuses any database whose +/// name does not match the per-test wrapper's (`buzz_nt_` plus 24 hex digits), +/// which excludes the run's `_desired` template. The name is the only check: +/// a manually created database with that shape would pass. +pub(crate) async fn retire_foreign_key_delete_actions(pool: &sqlx::PgPool) { + let database: String = sqlx::query_scalar("SELECT current_database()") + .fetch_one(pool) + .await + .expect("read current database"); + assert!( + database + .strip_prefix("buzz_nt_") + .is_some_and(|hash| hash.len() == 24 + && hash + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))), + "refusing to rewrite foreign keys in shared database {database}; \ + run this test through scripts/postgres-test-run.sh" + ); + sqlx::raw_sql( + r#"DO $$ + DECLARE c record; + BEGIN + FOR c IN SELECT conrelid::regclass AS tbl, conname, pg_get_constraintdef(oid) AS def + FROM pg_constraint + WHERE contype = 'f' AND confdeltype IN ('c', 'n', 'd') AND conparentid = 0 + LOOP + EXECUTE format('ALTER TABLE %s DROP CONSTRAINT %I, ADD CONSTRAINT %I %s', + c.tbl, c.conname, c.conname, + regexp_replace(c.def, ' ON DELETE (CASCADE|SET NULL|SET DEFAULT)( \([^)]*\))?', '')); + END LOOP; + END $$"#, + ) + .execute(pool) + .await + .expect("retire foreign-key delete actions"); + let remaining: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_constraint WHERE contype = 'f' AND confdeltype IN ('c', 'n', 'd')", + ) + .fetch_one(pool) + .await + .expect("count foreign-key delete actions"); + assert_eq!(remaining, 0, "every foreign-key delete action is retired"); +} + +/// A single-connection pool whose session carries `application_name`, so +/// [`wait_for_lock_wait`] can tell when its statement is blocked. +pub(crate) async fn named_pool(application_name: &str) -> sqlx::PgPool { + use std::str::FromStr as _; + let options = sqlx::postgres::PgConnectOptions::from_str(&database_url()) + .expect("parse test database URL") + .application_name(application_name); + sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .expect("connect named test pool") +} + +/// Wait until the session named `application_name` in this test's database is +/// blocked on a lock. +pub(crate) async fn wait_for_lock_wait(pool: &sqlx::PgPool, application_name: &str) { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity \ + WHERE datname = current_database() AND application_name = $1 \ + AND wait_event_type = 'Lock')", + ) + .bind(application_name) + .fetch_one(pool) + .await + .expect("inspect lock wait"); + if waiting { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("session reached a lock wait"); +} diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index bd39ada46e..b762af391d 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -8403,7 +8403,9 @@ pub(crate) mod postgres_tests { "CREATE SCHEMA {schema}; \ CREATE TABLE {schema}.users (LIKE public.users INCLUDING ALL); \ CREATE TABLE {schema}.community_bans (LIKE public.community_bans INCLUDING ALL); \ - CREATE TABLE {schema}.relay_members (LIKE public.relay_members INCLUDING ALL);" + CREATE TABLE {schema}.relay_members (LIKE public.relay_members INCLUDING ALL); \ + CREATE TABLE {schema}.join_policy_acceptances \ + (LIKE public.join_policy_acceptances INCLUDING ALL);" ))) .execute(state.db.pool()) .await