mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
fix(buzz-db): delete workflow and membership children explicitly (#8212)
## 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>
This commit is contained in:
Luke Tornquist
@@ -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()
|
||||
|
||||
@@ -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<bool> {
|
||||
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<RemoveResult> {
|
||||
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<RemoveResult> {
|
||||
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();
|
||||
|
||||
@@ -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<Option<Uuid>> {
|
||||
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<Option<Uuid>> {
|
||||
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");
|
||||
|
||||
|
||||
@@ -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<Uuid> = 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<Option<Uuid>> {
|
||||
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;
|
||||
|
||||
@@ -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<chrono::Utc> = 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}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user