From 5079c770fe30bb3d8204822ce6c2431eacac6d4b Mon Sep 17 00:00:00 2001 From: Wes Date: Mon, 21 Sep 2026 07:37:14 -0700 Subject: [PATCH] fix(workflows): make deletion persistent and retryable (#7735) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Fix workflow deletion leaving its definition queryable, and make public deletion acceptance agree atomically with executable/definition state. - Delete the executable workflow and soft-delete its definition under the existing coordinate lock, in the same transaction as kind-5 acceptance and mention indexing. Failed deletion leaves no public accepted event; identical retries can succeed. Legacy already-stored failed requests still repair. - Choose one committed dispatch/audit owner for concurrent identical requests. Completed duplicates remain silent. Preserve owner/community isolation, newer revisions, orphan cleanup, and intentional backdated recreation. - Align Desktop confirmation and architecture documentation with recreation semantics. Add deterministic PostgreSQL, live WebSocket, and Desktop regressions. No schema migration or new retry/delivery subsystem. ## Current validation Head: `40d86ddab0853ffb292316653b22c3fd4f983f0c`. - **Exact clean head:** live relay workflow-deletion E2E **4/4 passed**. All applicable normal pre-push hooks passed: Rust unit suites, full Desktop unit suite, Desktop lint/typecheck, Tauri clippy/tests, and file-size/branch checks. No hook bypass. - **Pre-commit implementation tree, then committed at this head:** full discovered DB/relay PostgreSQL lane **351/351 passed**; built Desktop deletion Playwright smoke **3/3 passed** (success, failure preserving editor/draft, corrected confirmation copy). - **Deterministic mutation evidence:** unchanged previous relay `659b79282` fails both new regressions: concurrent identical requests produce two normal/dispatch-owner responses, and rejected deletion leaves one public event. Both controls pass. Fixed relay passes all four. - **Independent source review:** transaction/authorization/lock-order and regression/cleanup lanes clear. Reviewers did not run tests independently. ## Honest limits / merge gates The broad pre-commit DB/relay/test-client selection passed **1,215/1,216**; the untouched mesh-demo echo test returned HTTP 504 instead of 200. It remains outside this repair and is not counted green. Initial hook attempts also exposed an untouched Desktop inventory ordering assertion and unidentified ACP/agent test failures; standalone package reruns and the final complete hooks passed without changing those packages. Causes are not established. The previous Desktop CI shard failed before tests on a Hermit `just` download (HTTP 500). New-head CI/security review and reviewer disposition remain required. No deployment, fresh interactive Desktop session, full local `just ci`, or multi-replica runtime validation claimed. Postgres/Redis validation used task-owned local containers, fresh identities, and disposable fixtures. ## Compatibility No automatic cleanup of existing workflows: repeating deletion repairs definitions left by the old path. Canonical UUID coordinates and legacy name fallback (`LIMIT 1` for duplicate names) are unchanged. Backdated recreation remains intentional per [Tyler’s decision](https://github.com/block/buzz/pull/7735#issuecomment-5729881295), not an irreversible tombstone. Mention-index failure now rolls back the complete deletion rather than best-effort accepting partial work. Detailed reviewer disposition and evidence: https://github.com/block/buzz/pull/7735#issuecomment-5737367825 --------- Signed-off-by: Wes Co-authored-by: Carl <32a2e2c9d428ee08902cab75d956da2c1d235a22d4766b0dd4138bf6e2e5db1d@buzz.block.builderlab.xyz> Co-authored-by: Mordecai --- ARCHITECTURE.md | 9 + crates/buzz-db/src/store/workflow.rs | 3 + crates/buzz-db/src/store/workflow/deletion.rs | 162 ++++ .../src/store/workflow/deletion_tests.rs | 294 +++++++ crates/buzz-relay/src/handlers/ingest.rs | 12 +- .../buzz-relay/src/handlers/side_effects.rs | 156 ++-- .../tests/e2e_workflow_deletion.rs | 780 ++++++++++++++++++ .../workflows/ui/WorkflowDeleteDialog.tsx | 2 +- desktop/tests/e2e/workflows.spec.ts | 9 +- 9 files changed, 1368 insertions(+), 59 deletions(-) create mode 100644 crates/buzz-db/src/store/workflow/deletion.rs create mode 100644 crates/buzz-db/src/store/workflow/deletion_tests.rs create mode 100644 crates/buzz-test-client/tests/e2e_workflow_deletion.rs 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(?:\?|$)/);