diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index bf435c270a..ae82c131ec 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -561,6 +561,15 @@ Note: Both `TriggerDef` and `ActionDef` use serde internally-tagged enums. Trigg **Cron scheduler:** loop ticks every 60 seconds, evaluates cron expressions with window-based matching, and creates workflow runs for matched triggers. Fully implemented. +**Deletion and recreation:** An authorized kind:5 `a`-tag deletion commits its +public event, executable workflow removal, and visible kind:30620 removal in one +transaction. Rejected deletions are not accepted history; identical concurrent +requests have one dispatch/audit owner. Deletions older than a live definition +leave that newer version intact. Deletion is not a permanent coordinate ban: +clients may publish a distinct signed definition afterward, including a backdated +version, to recreate the workflow and enable triggers again. This intentionally +retains arrival-order recreation without a deletion watermark. + **Does NOT:** recursively resolve templates (single-pass only). Does NOT queue workflow runs when at capacity — returns `CapacityExceeded` immediately. --- diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 3ceed9ea32..043ab112cf 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -7,6 +7,9 @@ //! - Approval tokens are stored as SHA-256 hashes (never plaintext). //! - All list queries have a bounded LIMIT to prevent unbounded scans. +mod deletion; +pub use deletion::WorkflowDeletionOutcome; + use std::fmt; use std::str::FromStr; diff --git a/crates/buzz-db/src/store/workflow/deletion.rs b/crates/buzz-db/src/store/workflow/deletion.rs new file mode 100644 index 0000000000..f9b8273566 --- /dev/null +++ b/crates/buzz-db/src/store/workflow/deletion.rs @@ -0,0 +1,162 @@ +//! Keep the executable workflow and its client-visible definition in sync. + +use buzz_core::{kind::KIND_WORKFLOW_DEF, CommunityId, StoredEvent}; +use buzz_datastore_tracing::datastore_span; +use chrono::{DateTime, Utc}; +use nostr::Event; +use sqlx::{Postgres, Row, Transaction}; +use uuid::Uuid; + +use crate::{Db, DbError, Result}; + +/// Committed changes made by a workflow-coordinate deletion. +#[derive(Debug, Default)] +pub struct WorkflowDeletionOutcome { + /// Whether either the executable workflow or a visible definition was removed. + pub changed: bool, + /// Removed executable workflow's channel, for trigger-cache invalidation. + pub channel_id: Option, +} + +impl Db { + /// Delete an authorized workflow coordinate and its definition atomically. + /// + /// Shares the replacement lock with definition saves. A deletion older than + /// the live definition is a no-op. Missing projections are tolerated so a + /// retry can remove definitions left behind by older relay versions. + /// Reports committed changes independently of the optional workflow channel. + /// No tombstone is retained: clients may intentionally publish backdated definitions. + #[datastore_span(name = "delete_workflow_by_coordinate", system = "postgresql")] + pub async fn delete_workflow_by_coordinate( + &self, + community_id: CommunityId, + owner_pubkey: &[u8], + d_tag: &str, + deletion_created_at_secs: i64, + ) -> Result { + let mut tx = self.begin_event_write_transaction().await?; + let outcome = delete_workflow_in_transaction( + &mut tx, + community_id, + owner_pubkey, + d_tag, + deletion_created_at_secs, + ) + .await?; + tx.commit().await?; + Ok(outcome) + } + + /// Commit the public deletion request and both workflow projections together. + /// + /// The returned boolean grants dispatch ownership: either this transaction + /// inserted the request, or it repaired an older relay's incomplete deletion. + /// Concurrent identical requests cannot split those two outcomes across commits. + /// Authorization of the coordinate is the caller's responsibility. + pub async fn insert_workflow_deletion( + &self, + community_id: CommunityId, + event: &Event, + owner_pubkey: &[u8], + d_tag: &str, + ) -> Result<(StoredEvent, bool, Option)> { + let mut tx = self.begin_event_write_transaction().await?; + let (stored, inserted) = + crate::event::insert_event_in_transaction(&mut tx, community_id, event, None).await?; + if inserted { + // Unlike best-effort indexing for ordinary events, deletion fails + // closed: its public request and discoverability commit together. + crate::runtime::insert_mentions_in_transaction(&mut tx, community_id, event, None) + .await?; + } + let outcome = delete_workflow_in_transaction( + &mut tx, + community_id, + owner_pubkey, + d_tag, + event.created_at.as_secs() as i64, + ) + .await?; + tx.commit().await?; + Ok((stored, inserted || outcome.changed, outcome.channel_id)) + } +} + +async fn delete_workflow_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + owner_pubkey: &[u8], + d_tag: &str, + deletion_created_at_secs: i64, +) -> Result { + let cutoff = DateTime::from_timestamp(deletion_created_at_secs, 0) + .ok_or(DbError::InvalidTimestamp(deletion_created_at_secs))?; + let lock_key = crate::store::replaceable::event_replacement_lock_key( + community_id, + KIND_WORKFLOW_DEF as i32, + owner_pubkey, + Some(d_tag.as_bytes()), + ); + crate::observability::observe_advisory_lock( + crate::observability::LockType::Replacement, + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(&mut **tx), + ) + .await?; + + let head: Option> = sqlx::query_scalar( + "SELECT created_at FROM events WHERE community_id = $1 AND kind = $2 \ + AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \ + ORDER BY created_at DESC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(owner_pubkey) + .bind(d_tag) + .fetch_optional(&mut **tx) + .await?; + if head.is_some_and(|created_at| created_at > cutoff) { + return Ok(WorkflowDeletionOutcome::default()); + } + + // UUID coordinates are canonical; retain the legacy name-based path. + // The owner predicate remains in the mutation, not just a prior check. + let workflow_id = Uuid::parse_str(d_tag).ok(); + let row = sqlx::query( + "DELETE 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", + ) + .bind(community_id.as_uuid()) + .bind(owner_pubkey) + .bind(workflow_id) + .bind(d_tag) + .fetch_optional(&mut **tx) + .await?; + 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", + ) + .bind(community_id.as_uuid()) + .bind(KIND_WORKFLOW_DEF as i32) + .bind(owner_pubkey) + .bind(d_tag) + .bind(cutoff) + .execute(&mut **tx) + .await?; + let changed = row.is_some() || definitions.rows_affected() > 0; + let channel_id = row + .map(|row| row.try_get("channel_id")) + .transpose()? + .flatten(); + Ok(WorkflowDeletionOutcome { + changed, + channel_id, + }) +} + +#[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 new file mode 100644 index 0000000000..04aa90326b --- /dev/null +++ b/crates/buzz-db/src/store/workflow/deletion_tests.rs @@ -0,0 +1,294 @@ +use super::*; +use crate::event::EventQuery; +use crate::workflow::{create_workflow_run, get_workflow, upsert_workflow}; +use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + +async fn setup() -> (Db, CommunityId) { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect test database"); + let db = Db::from_pool(pool); + let community = db + .ensure_configured_community(&format!("workflow-delete-{}.example", Uuid::new_v4())) + .await + .expect("create community") + .id; + (db, community) +} + +async fn seed( + db: &Db, + community: CommunityId, + keys: &Keys, + workflow_id: Uuid, + d_tag: &str, + created_at: u64, +) -> EventQuery { + let owner = keys.public_key().to_bytes(); + crate::user::ensure_user(&db.pool, community, &owner) + .await + .expect("owner"); + upsert_workflow( + &db.pool, + community, + workflow_id, + None, + &owner, + d_tag, + r#"{"trigger":{"on":"manual"},"steps":[]}"#, + &[0; 32], + ) + .await + .expect("seed executable workflow"); + let event = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + "name: deletion-test", + ) + .tags([Tag::parse(["d", d_tag]).expect("d tag")]) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(keys) + .expect("sign definition"); + db.replace_parameterized_event(community, &event, d_tag, None) + .await + .expect("seed definition"); + EventQuery { + kinds: Some(vec![KIND_WORKFLOW_DEF as i32]), + pubkey: Some(owner.to_vec()), + d_tag: Some(d_tag.to_string()), + ..EventQuery::for_community(community) + } +} + +async fn assert_present(db: &Db, query: &EventQuery, id: Uuid) { + assert_eq!( + db.query_events(query) + .await + .expect("fresh definition query") + .len(), + 1 + ); + assert!(get_workflow(&db.pool, query.community_id, id).await.is_ok()); +} + +async fn assert_absent(db: &Db, query: &EventQuery, id: Uuid) { + assert!(db + .query_events(query) + .await + .expect("fresh definition query") + .is_empty()); + assert!(matches!( + get_workflow(&db.pool, query.community_id, id).await, + Err(DbError::NotFound(_)) + )); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_deletion_removes_both_projections_and_is_scoped_and_retryable() { + let (db, community) = setup().await; + let other = db + .ensure_configured_community(&format!("other-{}.example", Uuid::new_v4())) + .await + .expect("other community") + .id; + 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_query = seed(&db, other, &keys, id, &d_tag, now).await; + assert_present(&db, &query, id).await; + + let unchanged = db + .delete_workflow_by_coordinate( + community, + &Keys::generate().public_key().to_bytes(), + &d_tag, + now as i64, + ) + .await + .expect("wrong owner is a no-op"); + assert!(!unchanged.changed); + assert_present(&db, &query, id).await; + + let run_id = create_workflow_run(&db.pool, community, id, None, None) + .await + .expect("run"); + // A fired schedule links both the workflow and a cascading run. Deletion + // must also remove the claim despite its NO ACTION run foreign key. + 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"); + let (first, second) = tokio::join!( + db.delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64), + db.delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64), + ); + let first = first.expect("first deletion"); + let second = second.expect("concurrent deletion"); + assert_ne!( + first.changed, second.changed, + "only one deletion changes state" + ); + assert_eq!( + first.channel_id, None, + "channel-less mutation still reports change" + ); + assert_eq!(second.channel_id, None); + assert_absent(&db, &query, id).await; + assert_present(&db, &other_query, id).await; + assert!(matches!( + db.get_workflow_run(community, run_id).await, + Err(DbError::NotFound(_)) + )); + let unchanged = db + .delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64) + .await + .expect("repeat deletion"); + assert!(!unchanged.changed); + assert_absent(&db, &query, id).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_deletion_preserves_newer_definition_and_cleans_legacy_orphan() { + let (db, community) = setup().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 unchanged = db + .delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64 - 1) + .await + .expect("stale deletion"); + assert!(!unchanged.changed); + 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) + .await + .expect("legacy deletion"); + assert_eq!( + db.query_events(&query) + .await + .expect("orphan definition") + .len(), + 1 + ); + let repaired = db + .delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64) + .await + .expect("repair deletion"); + assert!(repaired.changed, "orphan-only cleanup must dispatch"); + assert_eq!(repaired.channel_id, None); + assert_absent(&db, &query, id).await; + + let legacy_id = Uuid::new_v4(); + let legacy = seed(&db, community, &keys, legacy_id, "legacy-name", now).await; + db.soft_delete_by_coordinate( + community, + KIND_WORKFLOW_DEF as i32, + &owner, + "legacy-name", + now as i64, + ) + .await + .expect("remove definition to leave only executable projection"); + let repaired = db + .delete_workflow_by_coordinate(community, &owner, "legacy-name", now as i64) + .await + .expect("name coordinate deletion"); + assert!(repaired.changed, "projection-only cleanup must dispatch"); + assert_absent(&db, &legacy, legacy_id).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_deletion_rolls_back_both_representations_on_failure() { + let (db, community) = setup().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; + // A failing event update simulates a failure after projection removal. + // The CI runner isolates this test's database from other tests. + sqlx::raw_sql( + "CREATE FUNCTION reject_workflow_definition_delete() RETURNS trigger \ + LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected deletion failure'; END $$; \ + CREATE TRIGGER reject_workflow_definition_delete BEFORE UPDATE ON events \ + FOR EACH ROW WHEN (OLD.kind = 30620 AND NEW.deleted_at IS NOT NULL) \ + EXECUTE FUNCTION reject_workflow_definition_delete();", + ) + .execute(&db.pool) + .await + .expect("install failure trigger"); + let result = db + .delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64) + .await; + sqlx::raw_sql( + "DROP TRIGGER reject_workflow_definition_delete ON events; \ + DROP FUNCTION reject_workflow_definition_delete();", + ) + .execute(&db.pool) + .await + .expect("remove failure trigger"); + assert!(result.is_err()); + assert_present(&db, &query, id).await; + db.delete_workflow_by_coordinate(community, &owner, &d_tag, now as i64) + .await + .expect("retry after failure"); + assert_absent(&db, &query, id).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn legacy_stored_deletion_repair_has_one_committed_dispatch_owner() { + let (db, community) = setup().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 deletion = EventBuilder::new(Kind::EventDeletion, "") + .tags([Tag::parse([ + "a", + &format!("30620:{}:{d_tag}", keys.public_key().to_hex()), + ]) + .expect("coordinate")]) + .custom_created_at(Timestamp::from(now)) + .sign_with_keys(&keys) + .expect("sign legacy deletion"); + // Older relay versions could persist the request without applying deletion. + db.insert_event_with_thread_metadata(community, &deletion, None, None) + .await + .expect("legacy request"); + let (first, second) = tokio::join!( + db.insert_workflow_deletion(community, &deletion, &owner, &d_tag), + db.insert_workflow_deletion(community, &deletion, &owner, &d_tag), + ); + let (_, first_dispatch, _) = first.expect("first repair"); + let (_, second_dispatch, _) = second.expect("concurrent repair"); + assert_ne!( + first_dispatch, second_dispatch, + "only one committed repair dispatches" + ); + assert_absent(&db, &query, id).await; + assert!( + !db.insert_workflow_deletion(community, &deletion, &owner, &d_tag) + .await + .expect("completed duplicate") + .1 + ); +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index ee1d0312be..7f343c2c10 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3144,7 +3144,15 @@ async fn ingest_event_inner( }); } - let (stored_event, was_inserted) = if buzz_core::kind::is_replaceable(kind_u32) { + let workflow_deletion = crate::handlers::side_effects::is_workflow_deletion(&event); + let (stored_event, was_inserted) = if workflow_deletion { + // A single commit owns public acceptance, domain mutation, and dispatch. + // Failure rolls everything back; identical concurrent requests cannot + // divide insertion and repair ownership between two relay workers. + crate::handlers::side_effects::persist_workflow_deletion(tenant, &event, state) + .await + .map_err(|e| IngestError::Internal(format!("error: workflow deletion failed: {e}")))? + } else if buzz_core::kind::is_replaceable(kind_u32) { // NIP-16 replaceable event — atomic replace with stale-write protection. // channel_id is None for global kinds (0, 1, 3) due to step 5b above. state @@ -3211,7 +3219,7 @@ async fn ingest_event_inner( }); } - if crate::handlers::side_effects::is_side_effect_kind(kind_u32) { + if !workflow_deletion && crate::handlers::side_effects::is_side_effect_kind(kind_u32) { if let Err(e) = crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state) .await diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 7282c91342..702a0b62ca 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -2089,6 +2089,56 @@ async fn handle_leave_request( // handle_reaction() removed — kind:7 reaction dedup and DB writes are now // handled inline in ingest_event() before storage (see ingest.rs step 20a). +/// Whether the standard deletion handler will process a workflow coordinate. +pub(crate) fn is_workflow_deletion(event: &Event) -> bool { + // Match authorization and dispatch: e-tags take precedence, otherwise only + // the first a-tag is authorized and processed. + event.kind == Kind::EventDeletion + && !has_e_tag(event) + && event + .tags + .iter() + .find(|tag| tag.kind().to_string() == "a") + .and_then(|tag| tag.content()) + .is_some_and(|value| { + value + .split(':') + .next() + .and_then(|kind| kind.parse::().ok()) + == Some(buzz_core::kind::KIND_WORKFLOW_DEF) + }) +} + +/// Persist an already-authorized workflow deletion and its domain changes atomically. +pub(crate) async fn persist_workflow_deletion( + tenant: &TenantContext, + event: &Event, + state: &Arc, +) -> anyhow::Result<(buzz_core::StoredEvent, bool)> { + let coordinate = event + .tags + .iter() + .find(|tag| tag.kind().to_string() == "a") + .and_then(|tag| tag.content()) + .ok_or_else(|| anyhow::anyhow!("missing workflow coordinate"))?; + let mut parts = coordinate.splitn(3, ':'); + let _kind = parts.next(); + let owner = hex::decode(parts.next().unwrap_or_default())?; + let d_tag = parts + .next() + .ok_or_else(|| anyhow::anyhow!("invalid workflow coordinate"))?; + let (stored, dispatch, channel_id) = state + .db + .insert_workflow_deletion(tenant.community(), event, &owner, d_tag) + .await?; + if let Some(channel_id) = channel_id { + state + .workflow_engine + .invalidate_channel_workflows(tenant.community(), channel_id); + } + Ok((stored, dispatch)) +} + /// Handle NIP-09 deletion via `a` tag (addressable/parameterized-replaceable events). /// Parses "kind:pubkey:d-tag" and deletes the corresponding DB record. async fn handle_a_tag_deletion( @@ -2112,7 +2162,6 @@ async fn handle_a_tag_deletion( .map_err(|_| anyhow::anyhow!("invalid kind in a-tag"))?; let pubkey_hex = parts[1]; let d_tag = parts[2]; - let actor_bytes = effective_message_author(event, &state.relay_keypair.public_key()); match kind_num { // kind:30350 revocation is exclusively a higher-generation inactive replacement. @@ -2120,60 +2169,12 @@ async fn handle_a_tag_deletion( tracing::debug!(d_tag, "NIP-09 deletion ignored for push lease"); } buzz_core::kind::KIND_WORKFLOW_DEF => { - // Try UUID first (workflow_id); fall back to name-based lookup. - if let Ok(wf_id) = uuid::Uuid::parse_str(d_tag) { - let channel_id = state - .db - .delete_workflow_for_owner(tenant.community(), wf_id, &actor_bytes) - .await - .map_err(|e| anyhow::anyhow!("failed to delete workflow {wf_id}: {e}"))?; - if let Some(channel_id) = channel_id { - state - .workflow_engine - .invalidate_channel_workflows(tenant.community(), channel_id); - } - tracing::info!(workflow_id = %wf_id, "Workflow deleted via NIP-09 a-tag (UUID)"); - } else { - // Name-based lookup - match state - .db - .find_workflow_by_owner_and_name(tenant.community(), &actor_bytes, d_tag) - .await - { - Ok(Some(wf)) => { - let channel_id = state - .db - .delete_workflow_for_owner(tenant.community(), wf.id, &actor_bytes) - .await - .map_err(|e| { - anyhow::anyhow!("failed to delete workflow {}: {e}", wf.id) - })?; - if let Some(channel_id) = channel_id { - state - .workflow_engine - .invalidate_channel_workflows(tenant.community(), channel_id); - } - tracing::info!(workflow_id = %wf.id, name = d_tag, "Workflow deleted via NIP-09 a-tag (name)"); - } - Ok(None) => { - tracing::warn!( - "NIP-09 a-tag deletion: no workflow '{d_tag}' found for owner" - ); - } - Err(e) => { - tracing::warn!("NIP-09 a-tag deletion: DB lookup failed: {e}"); - } - } - } + // Workflow deletion belongs to the atomic persistence path in ingest. + return Err(anyhow::anyhow!( + "workflow deletion requires atomic persistence" + )); } - // Generic NIP-33 (parameterized-replaceable) soft-delete by coordinate. - // - // Listed after the workflow branch so workflow's bespoke deletion - // (which doesn't soft-delete the `events` row by design — that's a - // separate concern) takes precedence. For every other addressable - // kind, including kind:30023 (NIP-23 long-form), we soft-delete the - // live row matching `(kind, pubkey, d_tag)` so REQs stop returning it. - // See https://github.com/block/sprout/issues/714. + // Other NIP-33 events have no executable workflow projection. k if is_parameterized_replaceable(k) => { let pubkey_bytes = match hex::decode(pubkey_hex) { Ok(b) => b, @@ -3655,6 +3656,53 @@ pub async fn publish_nipia_unarchived( mod tests { use super::*; + #[test] + fn workflow_deletion_retry_matches_authorized_dispatch() { + let keys = nostr::Keys::generate(); + let workflow = format!("30620:{}:workflow", keys.public_key()); + let other = format!("30023:{}:article", keys.public_key()); + for (kind, tags, expected) in [ + ( + Kind::EventDeletion, + vec![vec!["a", workflow.as_str()]], + true, + ), + ( + Kind::EventDeletion, + vec![vec!["a", workflow.as_str()], vec!["e", "malformed"]], + false, + ), + ( + Kind::EventDeletion, + vec![vec!["a", other.as_str()], vec!["a", workflow.as_str()]], + false, + ), + ( + Kind::EventDeletion, + vec![vec!["a", workflow.as_str()], vec!["a", other.as_str()]], + true, + ), + ( + Kind::EventDeletion, + vec![vec!["a", "306200:owner:id"]], + false, + ), + ( + Kind::EventDeletion, + vec![vec!["a", "30620abc:owner:id"]], + false, + ), + (Kind::EventDeletion, vec![], false), + (Kind::TextNote, vec![vec!["a", workflow.as_str()]], false), + ] { + let event = EventBuilder::new(kind, "") + .tags(tags.into_iter().map(|tag| Tag::parse(tag).expect("tag"))) + .sign_with_keys(&keys) + .expect("sign"); + assert_eq!(is_workflow_deletion(&event), expected, "{:?}", event.tags); + } + } + #[test] fn nip43_reconciliation_compatibility_alias_is_preserved() { #[allow(deprecated)] diff --git a/crates/buzz-test-client/tests/e2e_workflow_deletion.rs b/crates/buzz-test-client/tests/e2e_workflow_deletion.rs new file mode 100644 index 0000000000..a46777307b --- /dev/null +++ b/crates/buzz-test-client/tests/e2e_workflow_deletion.rs @@ -0,0 +1,780 @@ +//! Relay-backed regression coverage for workflow deletion. +//! +//! # Running +//! +//! Start an isolated relay, then run: +//! +//! ```text +//! RELAY_URL=ws://localhost:3000 DATABASE_URL=postgres://… cargo test \ +//! -p buzz-test-client --test e2e_workflow_deletion -- --ignored +//! ``` + +use std::panic::AssertUnwindSafe; +use std::time::Duration; + +use buzz_test_client::{BuzzTestClient, RelayMessage, TestClientError}; +use futures_util::FutureExt; +use nostr::{Alphabet, Event, EventBuilder, Filter, Keys, Kind, SingleLetterTag, Tag, Timestamp}; +use sqlx::PgPool; +use uuid::Uuid; + +const KIND_WORKFLOW_DEFINITION: u16 = 30_620; +const KIND_WORKFLOW_TRIGGER: u16 = 46_020; + +fn relay_url() -> String { + std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()) +} + +fn database_url() -> String { + std::env::var("DATABASE_URL").expect("DATABASE_URL required for failure injection") +} + +fn sub_id(phase: &str) -> String { + format!("e2e-workflow-deletion-{phase}-{}", Uuid::new_v4()) +} + +fn definition_filter(keys: &Keys, workflow_id: &str) -> Filter { + Filter::new() + .kind(Kind::Custom(KIND_WORKFLOW_DEFINITION)) + .author(keys.public_key()) + .custom_tags(SingleLetterTag::lowercase(Alphabet::D), [workflow_id]) +} + +async fn query_definition( + client: &mut BuzzTestClient, + keys: &Keys, + workflow_id: &str, +) -> Vec { + let subscription = sub_id("definition-query"); + client + .subscribe(&subscription, vec![definition_filter(keys, workflow_id)]) + .await + .expect("subscribe for fresh workflow definition query"); + let events = client + .collect_until_eose(&subscription, Duration::from_secs(5)) + .await + .expect("query workflow definition"); + client + .close_subscription(&subscription) + .await + .expect("close definition query"); + events +} + +async fn query_deletion(client: &mut BuzzTestClient, deletion: &Event) -> Vec { + let subscription = sub_id("deletion-query"); + client + .subscribe( + &subscription, + vec![Filter::new().kind(Kind::EventDeletion).id(deletion.id)], + ) + .await + .expect("fresh exact-ID deletion query"); + let events = client + .collect_until_eose(&subscription, Duration::from_secs(5)) + .await + .expect("query deletion history"); + client + .close_subscription(&subscription) + .await + .expect("close deletion query"); + events +} + +async fn database_representation_counts( + pool: &PgPool, + keys: &Keys, + workflow_id: Uuid, +) -> (i64, i64) { + let owner = keys.public_key().to_bytes().to_vec(); + let d_tag = workflow_id.to_string(); + let workflow_rows: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM workflows WHERE id = $1 AND owner_pubkey = $2") + .bind(workflow_id) + .bind(&owner) + .fetch_one(pool) + .await + .expect("count executable workflow representation"); + let definition_rows: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM events WHERE kind = $1 AND pubkey = $2 AND d_tag = $3 \ + AND deleted_at IS NULL", + ) + .bind(i32::from(KIND_WORKFLOW_DEFINITION)) + .bind(&owner) + .bind(d_tag) + .fetch_one(pool) + .await + .expect("count visible workflow definition representation"); + (workflow_rows, definition_rows) +} + +struct FailureInjector { + pool: PgPool, +} + +impl FailureInjector { + async fn install(pool: PgPool, keys: &Keys, workflow_id: Uuid) -> Self { + // This fixture runs only against a dedicated disposable database. The + // shared trigger is inert unless its private table contains this test's + // exact owner + d-tag coordinate. + sqlx::raw_sql( + "CREATE TABLE workflow_delete_failure_injection (\ + owner_pubkey BYTEA NOT NULL, d_tag TEXT NOT NULL, \ + PRIMARY KEY (owner_pubkey, d_tag)\ + ); \ + CREATE FUNCTION reject_injected_workflow_delete() RETURNS trigger \ + LANGUAGE plpgsql AS $$ \ + BEGIN \ + IF OLD.kind = 30620 AND OLD.deleted_at IS NULL \ + AND NEW.deleted_at IS NOT NULL \ + AND EXISTS (SELECT 1 FROM workflow_delete_failure_injection \ + WHERE owner_pubkey = OLD.pubkey AND d_tag = OLD.d_tag) \ + THEN RAISE EXCEPTION 'injected workflow deletion failure'; \ + END IF; \ + RETURN NEW; \ + END $$; \ + CREATE TRIGGER reject_injected_workflow_delete \ + BEFORE UPDATE ON events FOR EACH ROW \ + EXECUTE FUNCTION reject_injected_workflow_delete();", + ) + .execute(&pool) + .await + .expect("install coordinate-scoped deletion failure injector"); + sqlx::query( + "INSERT INTO workflow_delete_failure_injection (owner_pubkey, d_tag) VALUES ($1, $2)", + ) + .bind(keys.public_key().to_bytes().to_vec()) + .bind(workflow_id.to_string()) + .execute(&pool) + .await + .expect("arm deletion failure injector for test coordinate"); + Self { pool } + } + + async fn remove(&mut self) { + sqlx::raw_sql( + "DROP TRIGGER IF EXISTS reject_injected_workflow_delete ON events; \ + DROP FUNCTION IF EXISTS reject_injected_workflow_delete(); \ + DROP TABLE IF EXISTS workflow_delete_failure_injection;", + ) + .execute(&self.pool) + .await + .expect("remove deletion failure injector"); + } +} + +async fn create_channel(client: &mut BuzzTestClient, keys: &Keys, channel_id: &str) { + let create_channel = EventBuilder::new(Kind::Custom(9007), "") + .tags([ + Tag::parse(["h", channel_id]).expect("channel h tag"), + Tag::parse(["name", "workflow-deletion-e2e"]).expect("channel name tag"), + Tag::parse(["channel_type", "stream"]).expect("channel type tag"), + Tag::parse(["visibility", "open"]).expect("channel visibility tag"), + ]) + .sign_with_keys(keys) + .expect("sign channel creation"); + let created = client + .send_event(create_channel) + .await + .expect("create channel"); + assert!( + created.accepted, + "channel creation rejected: {}", + created.message + ); +} + +fn workflow_definition(channel_id: &str, workflow_id: &str) -> EventBuilder { + EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEFINITION), + "name: deletion-regression\ntrigger:\n on: message_posted\nsteps:\n - id: pause\n action: delay\n duration: 1s\n", + ) + .tags([ + Tag::parse(["d", workflow_id]).expect("workflow d tag"), + Tag::parse(["h", channel_id]).expect("workflow h tag"), + ]) +} + +async fn create_workflow( + client: &mut BuzzTestClient, + keys: &Keys, + channel_id: &str, + workflow_id: &str, +) -> nostr::EventId { + let definition = workflow_definition(channel_id, workflow_id) + .sign_with_keys(keys) + .expect("sign workflow definition"); + let definition_id = definition.id; + let created = client + .send_event(definition) + .await + .expect("create workflow"); + assert!( + created.accepted, + "workflow creation rejected: {}", + created.message + ); + definition_id +} + +fn workflow_deletion(keys: &Keys, workflow_id: &str) -> Event { + let coordinate = format!( + "{KIND_WORKFLOW_DEFINITION}:{}:{workflow_id}", + keys.public_key().to_hex() + ); + EventBuilder::new(Kind::EventDeletion, "") + .tags([Tag::parse(["a", coordinate.as_str()]).expect("workflow coordinate")]) + .sign_with_keys(keys) + .expect("sign workflow deletion") +} + +async fn assert_trigger_rejected(client: &mut BuzzTestClient, keys: &Keys, workflow_id: &str) { + let trigger = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER), "{}") + .tags([Tag::parse(["d", workflow_id]).expect("trigger d tag")]) + .sign_with_keys(keys) + .expect("sign workflow trigger"); + let triggered = client + .send_event(trigger) + .await + .expect("submit post-deletion trigger"); + assert!(!triggered.accepted, "deleted workflow remained triggerable"); + assert!( + triggered.message.contains("workflow not found"), + "unexpected post-deletion trigger rejection: {}", + triggered.message + ); +} + +async fn assert_no_live_deletion(listener: &mut BuzzTestClient) { + assert!( + matches!( + listener.recv_event(Duration::from_millis(300)).await, + Err(TestClientError::Timeout) + ), + "rejected/completed duplicate deletion must not be delivered" + ); +} + +async fn assert_live_event(listener: &mut BuzzTestClient, subscription: &str, expected: &Event) { + match listener + .recv_event(Duration::from_secs(5)) + .await + .expect("receive live event") + { + RelayMessage::Event { + subscription_id, + event, + } => { + assert_eq!(subscription_id, subscription); + assert_eq!(*event, *expected, "dispatch the exact signed event"); + } + other => panic!("expected live event, got {other:?}"), + } +} + +async fn audit_count(pool: &PgPool, event: &Event) -> i64 { + sqlx::query_scalar("SELECT COUNT(*) FROM audit_log WHERE object_id = $1 AND action = 'event_created' AND actor_pubkey = $2 AND detail->>'event_kind' = $3") + .bind(event.id.to_hex()) + .bind(event.pubkey.to_bytes().to_vec()) + .bind(event.kind.as_u16().to_string()) + .fetch_one(pool).await.expect("count event audit entries") +} + +async fn flush_audit(client: &mut BuzzTestClient, keys: &Keys, pool: &PgPool) -> Event { + // Audit enqueue is awaited before OK. A later marker in the same relay's + // FIFO audit queue gives a causal flush fence instead of an arbitrary sleep. + let marker = EventBuilder::new(Kind::TextNote, Uuid::new_v4().to_string()) + .sign_with_keys(keys) + .expect("sign audit fence"); + assert!( + client + .send_event(marker.clone()) + .await + .expect("publish audit fence") + .accepted + ); + tokio::time::timeout(Duration::from_secs(5), async { + while audit_count(pool, &marker).await == 0 { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("audit worker must persist fence"); + marker +} + +#[tokio::test] +#[ignore] +async fn deleting_workflow_removes_definition_and_rejects_manual_trigger() { + let keys = Keys::generate(); + let channel_id = Uuid::new_v4().to_string(); + let workflow_id = Uuid::new_v4().to_string(); + let mut client = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("connect"); + + create_channel(&mut client, &keys, &channel_id).await; + let definition_id = create_workflow(&mut client, &keys, &channel_id, &workflow_id).await; + + let before = query_definition(&mut client, &keys, &workflow_id).await; + assert_eq!( + before.len(), + 1, + "definition must be queryable before deletion" + ); + assert_eq!(before[0].id, definition_id); + + let deletion = workflow_deletion(&keys, &workflow_id); + let deleted = client.send_event(deletion).await.expect("delete workflow"); + assert!( + deleted.accepted, + "workflow deletion rejected: {}", + deleted.message + ); + + let after = query_definition(&mut client, &keys, &workflow_id).await; + assert!( + after.is_empty(), + "deleted workflow definition remained queryable: {after:?}" + ); + + assert_trigger_rejected(&mut client, &keys, &workflow_id).await; + + client.disconnect().await.expect("disconnect"); +} + +async fn run_failed_deletion_replay_scenario(pool: &PgPool) { + let keys = Keys::generate(); + let channel_id = Uuid::new_v4().to_string(); + let workflow_id = Uuid::new_v4(); + let workflow_id_text = workflow_id.to_string(); + let mut client = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("connect"); + + create_channel(&mut client, &keys, &channel_id).await; + create_workflow(&mut client, &keys, &channel_id, &workflow_id_text).await; + assert_eq!( + database_representation_counts(pool, &keys, workflow_id).await, + (1, 1), + "workflow setup must create both representations" + ); + + let deletion = workflow_deletion(&keys, &workflow_id_text); + let deletion_id = deletion.id; + let mut listener = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("connect live listener"); + let subscription = sub_id("live-deletion"); + listener + .subscribe( + &subscription, + vec![Filter::new() + .kinds([Kind::EventDeletion, Kind::TextNote]) + .author(keys.public_key())], + ) + .await + .expect("subscribe before failure"); + assert!(listener + .collect_until_eose(&subscription, Duration::from_secs(5)) + .await + .expect("establish live subscription") + .is_empty()); + let mut injector = FailureInjector::install(pool.clone(), &keys, workflow_id).await; + let rejected = client + .send_event(deletion.clone()) + .await + .expect("submit injected-failure deletion"); + assert!( + !rejected.accepted, + "injected deletion failure was acknowledged" + ); + // WebSocket ingestion deliberately redacts internal database errors. + assert_eq!(rejected.message, "error: internal server error"); + let marker = flush_audit(&mut client, &keys, pool).await; + assert_live_event(&mut listener, &subscription, &marker).await; + assert_no_live_deletion(&mut listener).await; + assert_eq!( + audit_count(pool, &deletion).await, + 0, + "failed deletion must not be audited as accepted" + ); + + assert_eq!( + database_representation_counts(pool, &keys, workflow_id).await, + (1, 1), + "failed deletion must retain both workflow representations" + ); + let after_failure = query_definition(&mut client, &keys, &workflow_id_text).await; + assert_eq!( + after_failure.len(), + 1, + "definition disappeared after rejected deletion" + ); + + let stored_requests: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE id = $1") + .bind(deletion_id.to_bytes().to_vec()) + .fetch_one(pool) + .await + .expect("count stored deletion"); + assert_eq!( + stored_requests, 0, + "rejected deletion must roll back its public event" + ); + assert!( + query_deletion(&mut client, &deletion).await.is_empty(), + "rejected deletion must not appear in fresh REQ history" + ); + injector.remove().await; + let replayed = client + .send_event(deletion.clone()) + .await + .expect("replay identical signed deletion"); + assert!( + replayed.accepted, + "identical deletion replay rejected: {}", + replayed.message + ); + assert_eq!(replayed.event_id, deletion_id.to_hex()); + assert_eq!( + query_deletion(&mut client, &deletion).await, + vec![deletion.clone()] + ); + assert_live_event(&mut listener, &subscription, &deletion).await; + let marker = flush_audit(&mut client, &keys, pool).await; + assert_live_event(&mut listener, &subscription, &marker).await; + assert_eq!( + audit_count(pool, &deletion).await, + 1, + "repair must enqueue normal audit exactly once" + ); + assert_eq!( + database_representation_counts(pool, &keys, workflow_id).await, + (0, 0), + "successful replay must remove both workflow representations" + ); + assert!( + query_definition(&mut client, &keys, &workflow_id_text) + .await + .is_empty(), + "definition remained queryable after successful replay" + ); + assert_trigger_rejected(&mut client, &keys, &workflow_id_text).await; + let duplicate = client + .send_event(deletion.clone()) + .await + .expect("repeat completed deletion"); + assert!(duplicate.accepted && duplicate.message.starts_with("duplicate:")); + let marker = flush_audit(&mut client, &keys, pool).await; + assert_live_event(&mut listener, &subscription, &marker).await; + assert_no_live_deletion(&mut listener).await; + assert_eq!( + audit_count(pool, &deletion).await, + 1, + "completed duplicate must not re-audit" + ); + listener.disconnect().await.expect("disconnect listener"); + client.disconnect().await.expect("disconnect"); +} + +#[tokio::test] +#[ignore] +async fn failed_workflow_deletion_rolls_back_and_identical_event_can_be_replayed() { + let pool = PgPool::connect(&database_url()) + .await + .expect("connect to isolated workflow-deletion database"); + let result = AssertUnwindSafe(run_failed_deletion_replay_scenario(&pool)) + .catch_unwind() + .await; + + // Clean up even when the scenario panics so a failed run cannot poison the + // disposable relay for subsequent validation. + sqlx::raw_sql( + "DROP TRIGGER IF EXISTS reject_injected_workflow_delete ON events; \ + DROP FUNCTION IF EXISTS reject_injected_workflow_delete(); \ + DROP TABLE IF EXISTS workflow_delete_failure_injection;", + ) + .execute(&pool) + .await + .expect("final deletion failure injector cleanup"); + + if let Err(panic) = result { + std::panic::resume_unwind(panic); + } +} + +#[tokio::test] +#[ignore] +async fn workflow_can_be_intentionally_recreated_with_a_backdated_definition() { + let keys = Keys::generate(); + let channel_id = Uuid::new_v4().to_string(); + let workflow_id = Uuid::new_v4(); + let workflow_id_text = workflow_id.to_string(); + let mut client = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("connect"); + create_channel(&mut client, &keys, &channel_id).await; + create_workflow(&mut client, &keys, &channel_id, &workflow_id_text).await; + let deletion = workflow_deletion(&keys, &workflow_id_text); + let backdated = workflow_definition(&channel_id, &workflow_id_text) + .custom_created_at(Timestamp::from(deletion.created_at.as_secs() - 60)) + .sign_with_keys(&keys) + .expect("sign intentional backdated recreation"); + assert!(client.send_event(deletion).await.expect("delete").accepted); + assert!(query_definition(&mut client, &keys, &workflow_id_text) + .await + .is_empty()); + let recreated = client + .send_event(backdated.clone()) + .await + .expect("recreate backdated workflow"); + assert!( + recreated.accepted, + "backdating is intentional: {}", + recreated.message + ); + let definitions = query_definition(&mut client, &keys, &workflow_id_text).await; + assert_eq!(definitions.len(), 1); + assert_eq!(definitions[0].id, backdated.id); + let pool = PgPool::connect(&database_url()) + .await + .expect("isolated database"); + assert_eq!( + database_representation_counts(&pool, &keys, workflow_id).await, + (1, 1) + ); + let trigger = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER), "{}") + .tags([Tag::parse(["d", workflow_id_text.as_str()]).expect("trigger coordinate")]) + .sign_with_keys(&keys) + .expect("sign recreated workflow trigger"); + assert!( + client + .send_event(trigger) + .await + .expect("trigger recreated workflow") + .accepted + ); + client.disconnect().await.expect("disconnect"); +} + +async fn concurrent_deletion_scenario( + pool: &PgPool, + barrier: &mut Option>, + tasks: &mut tokio::task::JoinSet>, +) { + let keys = Keys::generate(); + let channel_id = Uuid::new_v4().to_string(); + let workflow_id = Uuid::new_v4(); + let workflow_id_text = workflow_id.to_string(); + let mut client = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("connect"); + create_channel(&mut client, &keys, &channel_id).await; + create_workflow(&mut client, &keys, &channel_id, &workflow_id_text).await; + let deletion = EventBuilder::new(Kind::EventDeletion, "") + .tags([ + Tag::parse([ + "a", + &format!("30620:{}:{workflow_id}", keys.public_key().to_hex()), + ]) + .expect("coordinate"), + Tag::public_key(Keys::generate().public_key()), + ]) + .sign_with_keys(&keys) + .expect("sign deletion"); + let mut listener = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("listener"); + let subscription = sub_id("concurrent-listener"); + listener + .subscribe( + &subscription, + vec![Filter::new() + .kinds([Kind::EventDeletion, Kind::TextNote]) + .author(keys.public_key())], + ) + .await + .expect("subscribe before concurrent requests"); + assert!(listener + .collect_until_eose(&subscription, Duration::from_secs(5)) + .await + .expect("listener EOSE") + .is_empty()); + + // Block A's first-insert-only mention indexing. In the old split path its + // public event is already committed and B can repair while A is held. In + // the atomic path B must wait on A's uncommitted unique event insertion. + // Both orderings are causally observed in Postgres, never guessed by sleep. + let lock_key = 7_735_306_205_i64; + *barrier = Some(pool.begin().await.expect("barrier transaction")); + let held = barrier.as_mut().expect("barrier exists"); + let controller: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut **held) + .await + .expect("controller pid"); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(&mut **held) + .await + .expect("hold indexing barrier"); + // Interpolated values are a fixed integer and an EventId's hex encoding, + // neither of which can contain SQL syntax. + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION block_workflow_deletion_index() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN IF NEW.event_id = decode('{}', 'hex') THEN + PERFORM pg_advisory_xact_lock({lock_key}); + END IF; RETURN NEW; END $$; + CREATE TRIGGER block_workflow_deletion_index BEFORE INSERT ON event_mentions + FOR EACH ROW EXECUTE FUNCTION block_workflow_deletion_index();", + deletion.id.to_hex() + ))) + .execute(pool) + .await + .expect("install event-scoped indexing barrier"); + + let mut second_client = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("second publisher"); + let first_event = deletion.clone(); + tasks.spawn(async move { client.send_event(first_event).await }); + let first_pid: i32 = + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let pid = sqlx::query_scalar::<_, i32>( + "SELECT pid FROM pg_stat_activity WHERE $1 = ANY(pg_blocking_pids(pid)) LIMIT 1") + .bind(controller).fetch_optional(pool).await.expect("find held first ingest"); + if let Some(pid) = pid { + break pid; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("first ingest must reach indexing barrier"); + let second_event = deletion.clone(); + let second = tasks.spawn(async move { second_client.send_event(second_event).await }); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let blocked: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE $1 = ANY(pg_blocking_pids(pid)))") + .bind(first_pid).fetch_one(pool).await.expect("observe second ingest wait"); + if blocked || second.is_finished() { break; } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await.expect("second ingest must either block on A or finish its repair"); + barrier + .take() + .expect("held barrier") + .commit() + .await + .expect("release first ingest"); + let first = tasks + .join_next() + .await + .expect("first task") + .expect("join first") + .expect("first response"); + let second = tasks + .join_next() + .await + .expect("second task") + .expect("join second") + .expect("second response"); + assert!( + first.accepted && second.accepted, + "both requests accepted: {first:?}, {second:?}" + ); + let mut client = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("verification client"); + let marker = flush_audit(&mut client, &keys, pool).await; + assert_eq!( + audit_count(pool, &deletion).await, + 1, + "concurrent ingests must have one dispatch/audit owner" + ); + assert_eq!( + [&first, &second] + .iter() + .filter(|response| response.message.is_empty()) + .count(), + 1, + "exactly one ingest owns normal acceptance" + ); + assert_eq!( + [&first, &second] + .iter() + .filter(|response| response.message.starts_with("duplicate:")) + .count(), + 1, + "the other ingest must be a completed duplicate" + ); + // Fan-out tasks can complete out of order; require each exact event once. + let mut expected = vec![deletion.clone(), marker]; + for _ in 0..2 { + match listener + .recv_event(Duration::from_secs(5)) + .await + .expect("receive deletion/fence") + { + RelayMessage::Event { + subscription_id, + event, + } => { + assert_eq!(subscription_id, subscription); + let index = expected + .iter() + .position(|candidate| candidate == event.as_ref()) + .expect("one copy of each exact deletion/fence event"); + expected.remove(index); + } + other => panic!("expected deletion/fence, got {other:?}"), + } + } + assert_no_live_deletion(&mut listener).await; + assert_eq!(query_deletion(&mut client, &deletion).await, vec![deletion]); + assert_eq!( + database_representation_counts(pool, &keys, workflow_id).await, + (0, 0) + ); + assert!(query_definition(&mut client, &keys, &workflow_id_text) + .await + .is_empty()); + assert_trigger_rejected(&mut client, &keys, &workflow_id_text).await; +} + +#[tokio::test] +#[ignore] +async fn concurrent_identical_deletions_dispatch_and_audit_once() { + let pool = PgPool::connect(&database_url()) + .await + .expect("isolated database"); + let mut barrier = None; + let mut tasks = tokio::task::JoinSet::new(); + let result = AssertUnwindSafe(concurrent_deletion_scenario( + &pool, + &mut barrier, + &mut tasks, + )) + .catch_unwind() + .await; + // Release blocked server transactions before dropping a trigger they use. + if let Some(barrier) = barrier { + barrier + .rollback() + .await + .expect("release failed scenario barrier"); + } + tasks.abort_all(); + while tasks.join_next().await.is_some() {} + sqlx::raw_sql( + "DROP TRIGGER IF EXISTS block_workflow_deletion_index ON event_mentions; + DROP FUNCTION IF EXISTS block_workflow_deletion_index();", + ) + .execute(&pool) + .await + .expect("remove indexing barrier"); + if let Err(panic) = result { + std::panic::resume_unwind(panic); + } +} diff --git a/desktop/src/features/workflows/ui/WorkflowDeleteDialog.tsx b/desktop/src/features/workflows/ui/WorkflowDeleteDialog.tsx index 1c0b1aeb05..0e07c28a9b 100644 --- a/desktop/src/features/workflows/ui/WorkflowDeleteDialog.tsx +++ b/desktop/src/features/workflows/ui/WorkflowDeleteDialog.tsx @@ -39,7 +39,7 @@ export function WorkflowDeleteDialog({ Delete workflow? {workflow - ? `Delete "${workflow.name}". This will stop all future triggers and remove the workflow permanently.` + ? `Delete "${workflow.name}". This removes the current workflow and stops new triggers. Publishing its definition again, even an older version, can recreate it.` : "Delete this workflow."} {error ? ( diff --git a/desktop/tests/e2e/workflows.spec.ts b/desktop/tests/e2e/workflows.spec.ts index 16844a2f3e..baf7e29a9c 100644 --- a/desktop/tests/e2e/workflows.spec.ts +++ b/desktop/tests/e2e/workflows.spec.ts @@ -642,8 +642,13 @@ test("deleting the open workflow closes its editor", async ({ page }) => { await expect(detailDialog).toBeVisible(); await detailDialog.getByRole("button", { name: "Workflow actions" }).click(); await page.getByRole("menuitem", { name: "Delete" }).click(); - await expect(page.getByRole("alertdialog")).toContainText(workflowName); - await page.getByRole("button", { name: "Delete" }).click(); + const confirmation = page.getByRole("alertdialog"); + await expect(confirmation).toContainText(workflowName); + await expect(confirmation).toContainText( + "Publishing its definition again, even an older version, can recreate it.", + ); + await expect(confirmation).not.toContainText("permanently"); + await confirmation.getByRole("button", { name: "Delete" }).click(); await expect(detailDialog).toHaveCount(0); await expect(page).toHaveURL(/#\/workflows(?:\?|$)/);